@hile/reactivity

Compose Hile micro topics, dynamic config, and runtime reloader state as refs with explicit async watch strategies.

@hile/reactivity

Compose Hile micro topics, dynamic config, and runtime reloader state as refs with explicit async watch strategies.

Choose This Package When

User asks for	Use	Also read
Compose topics, config, and reloader state as refs	@hile/reactivity	packages/reactivity.md, recipes/stable-runtime-reload.md

Use When

Use @hile/reactivity when Hile service code needs to compose micro topics, dynamic config, or runtime reloader state as refs instead of manual callback chains.

Do Not Use When

• Do not use it to replace @hile/micro; it adapts publish() and subscribe() but does not change RPC or pub/sub semantics.
• Do not use it to replace @hile/reloader; use reactiveReloader() when callers need refs for current, state, and error.
• Do not use it as an event log. Refs model current state; use a queue for durable event processing.
• Do not use watchLatest for side effects that must all run. Use watchQueue for every-event handling.
• Do not rely on AbortSignal to kill non-cooperative async work. It prevents stale writes only when handlers respect the signal or check isCurrent().

Install

pnpm add @hile/reactivity

Imports

import {
  topicRef,
  publishRef,
  reactiveConfig,
  reactiveReloader,
  watchLatest,
  watchSerialLatest,
  watchQueue,
  shallowRef,
  computed,
} from '@hile/reactivity'

Copy-Paste Example

Reactive config topics feeding a reloadable runtime:

import { reactiveConfig, reactiveReloader, watchSerialLatest } from '@hile/reactivity'

type RuntimeConfig = {
  mysql: { host: string; port: number }
  redis: { host: string; port: number }
  flags: Record<string, boolean>
}

const configs = await reactiveConfig<RuntimeConfig>(app, {
  topics: {
    mysql: 'config:mysql',
    redis: 'config:redis',
    flags: 'config:flags',
  },
  required: ['mysql', 'redis'],
  defaults: { flags: {} },
})

const runtime = reactiveReloader<RuntimeConfig, Runtime>({
  debounceMs: 100,
  create: createRuntime,
  dispose: runtime => runtime.close(),
  onError: (error, context) => {
    logger.error({ error, stage: context.stage }, 'runtime reload failed')
  },
})

const stopReload = watchSerialLatest(configs.current, async config => {
  if (!config) return
  await runtime.update(config)
})

http.use(async (ctx, next) => {
  const handler = runtime.current.value
  if (!handler) {
    ctx.status = 503
    ctx.body = { error: 'runtime is not ready' }
    return
  }
  await handler.handle(ctx, next)
})

shutdown(async () => {
  stopReload()
  await configs.stop()
  await runtime.stop()
})

More Examples

Subscribe to one topic as a readonly ref:

import { topicRef, watchLatest } from '@hile/reactivity'

const limit = await topicRef<number>(app, 'config:limit', { defaultValue: 10 })

watchLatest(limit.ref, async value => {
  logger.info({ value }, 'limit changed')
})

shutdown(limit.stop)

Publish a local ref as a topic with serial latest-wins updates:

import { publishRef, shallowRef } from '@hile/reactivity'

const limit = shallowRef(10)
const published = await publishRef(app, 'runtime:limit', limit)

limit.value = 20

shutdown(published.stop)

Choose an async watch strategy explicitly:

import { watchLatest, watchSerialLatest, watchQueue } from '@hile/reactivity'

watchLatest(searchText, async (value, { signal }) => {
  const result = await fetchResult(value, { signal })
  cache.value = result
})

watchSerialLatest(configRef, async config => {
  await reloader.update(config)
})

watchQueue(auditEventRef, async event => {
  await appendAuditLog(event)
})

Runtime And Lifecycle Notes

• topicRef() subscribes once, stores the latest payload in a readonly ref, marks ready after the first payload, and unsubscribes on stop().
• publishRef() publishes the initial ref value, then watches later changes with serial latest-wins updates. stop() drops pending values that have not started, waits for the active publisher.update() to settle, then calls publisher.unpublish().
• reactiveConfig() uses createConfigAggregator() internally, so required keys, defaults, debounce, signatures, and listener error semantics stay aligned with @hile/reloader.
• reactiveConfig().stop() attempts every subscribed cleanup before reporting unsubscribe failures. Failed cleanup remains retryable, and onError can receive stage: 'unsubscribe'.
• reactiveReloader() wraps createRuntimeReloader() and exposes readonly refs for current, state, and error.
• watchLatest() aborts stale work and reports errors only for the latest non-aborted run.
• watchSerialLatest() never runs handlers concurrently and collapses pending changes to the latest value.
• watchQueue() never runs handlers concurrently and processes every observed value in order.
• @vue/reactivity is a peer dependency. Import refs and watch helpers from @hile/reactivity, or ensure the application resolves a single physical copy of @vue/reactivity.
• Register every returned stop() with Hile shutdown.

Anti-Patterns

• Starting async work inside a raw watch() without choosing latest-wins, serial latest-wins, or queue semantics.
• Updating a micro topic directly from multiple uncoordinated watchers.
• Creating refs from a second physical copy of @vue/reactivity; cross-instance refs can be recognized as refs but not tracked by this package’s watchers.
• Passing large mutable objects through deep reactive proxies. Prefer shallowRef for Hile runtime/config snapshots.
• Forgetting to stop topic subscriptions or publisher bindings during service shutdown.

Verification Checklist

• Unit-test async watcher behavior with overlapping updates.
• Verify topicRef().stop() unsubscribes and later topic pushes do not mutate the ref.
• Verify publishRef() skips superseded pending values and calls unpublish() on stop.
• Verify reactiveConfig() does not emit until required topics are present.
• Verify reactiveConfig() preserves subscribe failures even when best-effort cleanup also fails, and that failed stop cleanup can be retried.
• Verify reactiveReloader() preserves @hile/reloader failure behavior and exposes current runtime after successful update.

Related Recipes

Stable Runtime Reload

Complete Example

import { defineService, loadService } from '@hile/core'
import { Application } from '@hile/micro'
import { Http } from '@hile/http'
import { createConfigAggregator, createRuntimeReloader } from '@hile/reloader'
import appService from './micro.app.boot'
import httpService from './http.boot'

type RuntimeConfig = {
  mysql: { host: string; port: number }
  redis: { host: string; port: number }
  flags: Record<string, boolean>
}

export default defineService('runtime.reload', async (shutdown) => {
  const app = await loadService<Application>(appService)
  const http = await loadService<Http>(httpService)

  const reloader = createRuntimeReloader<RuntimeConfig, AppRuntime>({
    debounceMs: 100,
    normalize: config => ({
      mysql: config.mysql,
      redis: config.redis,
      flags: config.flags,
    }),
    create: config => createAppRuntime(config),
    dispose: runtime => runtime.close(),
    onError: (error, context) => {
      logger.error({ error, stage: context.stage }, 'runtime reload failed')
    },
  })

  const configs = createConfigAggregator<RuntimeConfig>({
    required: ['mysql', 'redis'],
    defaults: { flags: {} },
    debounceMs: 100,
    onError: (error) => {
      logger.error({ error }, 'runtime config emit failed')
    },
  })

  configs.onChange(config => reloader.update(config))

  const cleanupMysql = await app.subscribe('config:mysql', value => configs.set('mysql', value))
  const cleanupRedis = await app.subscribe('config:redis', value => configs.set('redis', value))
  const cleanupFlags = await app.subscribe('config:flags', value => configs.set('flags', value))

  http.use(async (ctx, next) => {
    const runtime = reloader.current()
    if (!runtime) {
      ctx.status = 503
      ctx.body = { error: 'runtime is not ready' }
      return
    }
    await runtime.handle(ctx, next)
  })

  shutdown(async () => {
    await cleanupMysql()
    await cleanupRedis()
    await cleanupFlags()
    configs.dispose()
    await reloader.stop()
  })

  return reloader
})

Runtime Dynamic Config

Complete Example

import { z } from 'zod'
import { MicroDynamicConfigsServer } from '@hile/micro-dynamic-configs'

const schema = z.object({
  featureCheckout: z.boolean().default(false),
  maxRetries: z.number().int().min(1).max(10).default(3),
})

const configs = new MicroDynamicConfigsServer({
  app,
  redis,
  schema,
  redis_key: 'configs:checkout',
})

const cleanup = await configs.initialize()
shutdown(cleanup)

configs.on('change:featureCheckout', (next, previous) => {
  logger.info({ previous, next }, 'featureCheckout changed')
})

await configs.save({ featureCheckout: true })

Package-Local AI Guide

This package also ships AI.md in npm so agents can read accurate examples after installation.