feat(metadata-core): LayeredRepository (ADR-0008 M0 PR-5)

- src/layered-repository.ts: composes N child repos into a read-through
  stack. Reads walk top-to-bottom; writes go to the topmost writable
  layer; list() deduplicates by refKey; history() and watch() merge
  events from all layers, tagging each event source as
  <layer>:<original-source>. Multiplexed watch correctly cancels all
  child iterators via return().
- test/layered-repository.test.ts: 8 tests cover shadowing, fall-through,
  write routing, no-writable-layer error, list dedup, source tagging,
  child-iterator cancellation, and history merge. 79/79 metadata-core
  tests pass.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: os-zhuang

.changeset/metadata-core-layered.md

@@ -0,0 +1,16 @@
+---
+'@objectstack/metadata-core': minor
+---
+
+Add `LayeredRepository` — composes N `MetadataRepository`s into a
+read-through stack. Reads walk top-to-bottom; writes route to the
+topmost writable layer; `list()` deduplicates by `refKey` preferring
+the top; `history()` and `watch()` merge events from all layers,
+tagging each event's `source` with `<layer>:<original-source>`. The
+multiplexed `watch()` correctly cancels all child iterators when the
+consumer calls `return()`.
+
+Enables the canonical "system built-ins under user overlay" pattern
+described in ADR-0008.
+
+See ADR-0008 §10 PR-5.

packages/metadata-core/src/index.ts

@@ -12,3 +12,4 @@ export * from './canonicalize.js';
 export * from './repository.js';
 export * from './in-memory-repository.js';
 export * from './cache.js';
+export * from './layered-repository.js';

packages/metadata-core/src/layered-repository.ts

@@ -0,0 +1,220 @@
+// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license.
+
+/**
+ * `LayeredRepository` — composes N child `MetadataRepository`s into a
+ * single read-through stack. See ADR-0008 §10 PR-5.
+ *
+ * Read semantics
+ * ──────────────
+ *   - `get(ref)` walks the layers top-to-bottom; first non-null wins.
+ *   - `list()` deduplicates by `refKey(ref)`, preferring the top layer.
+ *   - `history()` and `watch()` merge events from all layers, each
+ *     tagged with the source layer label in `evt.source`
+ *     (`<label>:<original-source>`).
+ *
+ * Write semantics
+ * ───────────────
+ *   - `put` / `delete` are routed to the topmost writable layer.
+ *   - A layer is writable unless `readOnly: true` in its config.
+ *   - The write target's existing HEAD is used for the parent check —
+ *     i.e. if a key exists only in a lower layer, writing produces a
+ *     create on the top layer (an "overlay" of the lower one).
+ *
+ * Event tagging
+ * ─────────────
+ *   - Each emitted event carries `source` rewritten as
+ *     `<layer.label>:<source>` so subscribers can tell which layer
+ *     produced the change. Original event content is otherwise
+ *     unchanged.
+ *
+ * This implementation never re-orders events relative to seq within a
+ * single layer, but seqs across layers may interleave. Downstream
+ * consumers (Cache, SchemaRegistry) only need monotonicity within a
+ * (layer, branch) tuple — they treat each (layer.label, ref) as a
+ * separate stream.
+ */
+
+import {
+  type MetadataRepository,
+  type MetaRef,
+  type MetadataItem,
+  type MetadataItemHeader,
+  type MetadataEvent,
+  type PutOptions,
+  type PutResult,
+  type DeleteOptions,
+  type DeleteResult,
+  type ListFilter,
+  type WatchFilter,
+  type HistoryOptions,
+  refKey,
+} from './index.js';
+
+export interface LayerConfig {
+  /** Stable identifier for this layer (e.g. "user", "team", "system"). */
+  label: string;
+  /** The backing repository. */
+  repo: MetadataRepository;
+  /** When true, this layer rejects writes (used for built-ins). */
+  readOnly?: boolean;
+}
+
+export interface LayeredRepositoryOptions {
+  /**
+   * Layers in priority order, highest first. The first writable layer
+   * receives all writes.
+   */
+  layers: LayerConfig[];
+}
+
+const tagSource = (label: string, evt: MetadataEvent): MetadataEvent => ({
+  ...evt,
+  source: `${label}:${evt.source}`,
+});
+
+export class LayeredRepository implements MetadataRepository {
+  private readonly layers: LayerConfig[];
+  /** Index of the first writable layer; -1 if none. */
+  private readonly writableIdx: number;
+
+  constructor(opts: LayeredRepositoryOptions) {
+    if (!opts.layers.length) {
+      throw new Error('LayeredRepository requires at least one layer');
+    }
+    this.layers = opts.layers;
+    this.writableIdx = opts.layers.findIndex((l) => !l.readOnly);
+  }
+
+  async get(ref: MetaRef): Promise<MetadataItem | null> {
+    for (const layer of this.layers) {
+      const item = await layer.repo.get(ref);
+      if (item) return item;
+    }
+    return null;
+  }
+
+  async put(ref: MetaRef, spec: unknown, opts: PutOptions): Promise<PutResult> {
+    if (this.writableIdx < 0) {
+      throw new Error('LayeredRepository: no writable layer configured');
+    }
+    return this.layers[this.writableIdx]!.repo.put(ref, spec, opts);
+  }
+
+  async delete(ref: MetaRef, opts: DeleteOptions): Promise<DeleteResult> {
+    if (this.writableIdx < 0) {
+      throw new Error('LayeredRepository: no writable layer configured');
+    }
+    return this.layers[this.writableIdx]!.repo.delete(ref, opts);
+  }
+
+  async *list(filter: ListFilter): AsyncIterable<MetadataItemHeader> {
+    // Yield headers from top→bottom, deduplicating by refKey.
+    const seen = new Set<string>();
+    const limit = filter.limit ?? Infinity;
+    let yielded = 0;
+    for (const layer of this.layers) {
+      for await (const header of layer.repo.list(filter)) {
+        const key = refKey(header.ref);
+        if (seen.has(key)) continue;
+        seen.add(key);
+        yield header;
+        if (++yielded >= limit) return;
+      }
+    }
+  }
+
+  async *history(ref: MetaRef, opts: HistoryOptions = {}): AsyncIterable<MetadataEvent> {
+    // Merge histories from all layers, tagged. Order preserves each
+    // layer's monotonic seq, but events across layers may interleave.
+    // We collect everything then sort by ts as a best-effort total order.
+    const events: MetadataEvent[] = [];
+    for (const layer of this.layers) {
+      for await (const evt of layer.repo.history(ref, opts)) {
+        events.push(tagSource(layer.label, evt));
+      }
+    }
+    events.sort((a, b) => (a.ts < b.ts ? -1 : a.ts > b.ts ? 1 : 0));
+    const limit = opts.limit ?? Infinity;
+    let yielded = 0;
+    for (const evt of events) {
+      yield evt;
+      if (++yielded >= limit) return;
+    }
+  }
+
+  watch(filter: WatchFilter, since?: number): AsyncIterable<MetadataEvent> {
+    // Fan out to all child watchers; multiplex into a single iterator.
+    return multiplexWatch(this.layers, filter, since);
+  }
+}
+
+/**
+ * Multiplex N async iterables of MetadataEvent into one, tagging each
+ * event's `source` with its layer label. Implemented as a manual
+ * AsyncIterator so we can correctly cancel all child iterators when the
+ * consumer breaks out.
+ */
+function multiplexWatch(
+  layers: LayerConfig[],
+  filter: WatchFilter,
+  since: number | undefined,
+): AsyncIterable<MetadataEvent> {
+  return {
+    [Symbol.asyncIterator]() {
+      const children = layers.map((layer) => ({
+        label: layer.label,
+        iter: layer.repo.watch(filter, since)[Symbol.asyncIterator](),
+        pending: null as Promise<{ label: string; result: IteratorResult<MetadataEvent> }> | null,
+        done: false,
+      }));
+      let closed = false;
+
+      const pumpAll = () => {
+        for (const c of children) {
+          if (c.done || c.pending) continue;
+          c.pending = c.iter.next().then((result) => ({ label: c.label, result }));
+        }
+      };
+
+      const closeAll = async () => {
+        if (closed) return;
+        closed = true;
+        await Promise.all(
+          children.map(async (c) => {
+            try { await c.iter.return?.(undefined); } catch { /* ignore */ }
+          }),
+        );
+      };
+
+      return {
+        async next(): Promise<IteratorResult<MetadataEvent>> {
+          while (!closed) {
+            pumpAll();
+            const inflight = children.filter((c) => c.pending);
+            if (!inflight.length) return { value: undefined, done: true };
+            const winner = await Promise.race(inflight.map((c) => c.pending!));
+            const target = children.find((c) => c.label === winner.label)!;
+            target.pending = null;
+            if (winner.result.done) {
+              target.done = true;
+              continue;
+            }
+            return {
+              value: tagSource(winner.label, winner.result.value as MetadataEvent),
+              done: false,
+            };
+          }
+          return { value: undefined, done: true };
+        },
+        async return() {
+          await closeAll();
+          return { value: undefined, done: true };
+        },
+        async throw(err) {
+          await closeAll();
+          throw err;
+        },
+      } as AsyncIterator<MetadataEvent>;
+    },
+  };
+}

packages/metadata-core/test/layered-repository.test.ts

@@ -0,0 +1,155 @@
+// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license.
+
+import { describe, it, expect } from 'vitest';
+import { InMemoryRepository } from '../src/in-memory-repository.js';
+import { LayeredRepository } from '../src/layered-repository.js';
+import type { MetaRef, MetadataEvent } from '../src/types.js';
+
+const ref = (name: string): MetaRef => ({
+  org: 'system',
+  project: 'test',
+  branch: 'main',
+  type: 'view',
+  name,
+});
+
+const sleep = (ms: number) => new Promise<void>((r) => setTimeout(r, ms));
+
+describe('LayeredRepository', () => {
+  it('get: top layer shadows lower layers', async () => {
+    const builtins = new InMemoryRepository();
+    const userspace = new InMemoryRepository();
+    await builtins.put(ref('a'), { from: 'builtin' }, { parentVersion: null, actor: 't' });
+    await userspace.put(ref('a'), { from: 'user' }, { parentVersion: null, actor: 't' });
+
+    const layered = new LayeredRepository({
+      layers: [
+        { label: 'user', repo: userspace },
+        { label: 'system', repo: builtins, readOnly: true },
+      ],
+    });
+    const got = await layered.get(ref('a'));
+    expect(got?.body).toEqual({ from: 'user' });
+  });
+
+  it('get: falls through to lower layers when top is empty', async () => {
+    const builtins = new InMemoryRepository();
+    const userspace = new InMemoryRepository();
+    await builtins.put(ref('only_in_builtin'), { v: 1 }, { parentVersion: null, actor: 't' });
+
+    const layered = new LayeredRepository({
+      layers: [
+        { label: 'user', repo: userspace },
+        { label: 'system', repo: builtins, readOnly: true },
+      ],
+    });
+    const got = await layered.get(ref('only_in_builtin'));
+    expect(got?.body).toEqual({ v: 1 });
+  });
+
+  it('put: routes to the topmost writable layer', async () => {
+    const builtins = new InMemoryRepository();
+    const userspace = new InMemoryRepository();
+    const layered = new LayeredRepository({
+      layers: [
+        { label: 'user', repo: userspace },
+        { label: 'system', repo: builtins, readOnly: true },
+      ],
+    });
+    await layered.put(ref('written'), { v: 1 }, { parentVersion: null, actor: 't' });
+    expect(await userspace.get(ref('written'))).not.toBeNull();
+    expect(await builtins.get(ref('written'))).toBeNull();
+  });
+
+  it('put: throws if no writable layer exists', async () => {
+    const ro = new InMemoryRepository();
+    const layered = new LayeredRepository({
+      layers: [{ label: 'system', repo: ro, readOnly: true }],
+    });
+    await expect(
+      layered.put(ref('x'), { v: 1 }, { parentVersion: null, actor: 't' }),
+    ).rejects.toThrow(/no writable layer/);
+  });
+
+  it('list: deduplicates by refKey, preferring the top layer', async () => {
+    const builtins = new InMemoryRepository();
+    const userspace = new InMemoryRepository();
+    await builtins.put(ref('shared'), { from: 'builtin' }, { parentVersion: null, actor: 't' });
+    await builtins.put(ref('only_builtin'), { v: 1 }, { parentVersion: null, actor: 't' });
+    await userspace.put(ref('shared'), { from: 'user' }, { parentVersion: null, actor: 't' });
+    await userspace.put(ref('only_user'), { v: 1 }, { parentVersion: null, actor: 't' });
+
+    const layered = new LayeredRepository({
+      layers: [
+        { label: 'user', repo: userspace },
+        { label: 'system', repo: builtins, readOnly: true },
+      ],
+    });
+    const names: string[] = [];
+    for await (const header of layered.list({ type: 'view' })) {
+      names.push(header.ref.name);
+    }
+    expect(new Set(names)).toEqual(new Set(['shared', 'only_user', 'only_builtin']));
+    // 'shared' must come from user (top) → headers should be unique.
+    expect(names.length).toBe(3);
+  });
+
+  it('watch: tags events with <layer>:<source>', async () => {
+    const a = new InMemoryRepository();
+    const b = new InMemoryRepository();
+    const layered = new LayeredRepository({
+      layers: [
+        { label: 'top', repo: a },
+        { label: 'bottom', repo: b },
+      ],
+    });
+    const iter = layered.watch({ org: 'system', project: 'test', branch: 'main' })[Symbol.asyncIterator]();
+    const collected: MetadataEvent[] = [];
+    const done = (async () => {
+      for (let i = 0; i < 2; i++) {
+        const r = await iter.next();
+        if (r.done) return;
+        collected.push(r.value as MetadataEvent);
+      }
+    })();
+    await sleep(10);
+    await a.put(ref('x'), { from: 'top' }, { parentVersion: null, actor: 't' });
+    await b.put(ref('y'), { from: 'bottom' }, { parentVersion: null, actor: 't' });
+    await Promise.race([done, sleep(1000)]);
+    await iter.return?.(undefined);
+    expect(collected).toHaveLength(2);
+    const sources = collected.map((e) => e.source).sort();
+    expect(sources).toEqual(['bottom:in-memory', 'top:in-memory']);
+  });
+
+  it('watch: closing the layered iterator closes all child iterators', async () => {
+    const a = new InMemoryRepository();
+    const b = new InMemoryRepository();
+    const layered = new LayeredRepository({
+      layers: [{ label: 'a', repo: a }, { label: 'b', repo: b }],
+    });
+    const iter = layered.watch({ org: 'system', project: 'test', branch: 'main' })[Symbol.asyncIterator]();
+    // Schedule a next() that will park (no events yet).
+    const pending = iter.next();
+    await sleep(10);
+    // Closing must unblock pending; otherwise this test will time out.
+    await iter.return?.(undefined);
+    const result = await pending;
+    expect(result.done).toBe(true);
+  });
+
+  it('history: merges from all layers, sorted by ts', async () => {
+    const a = new InMemoryRepository({ now: () => new Date('2025-01-01T00:00:00Z') });
+    const b = new InMemoryRepository({ now: () => new Date('2025-01-02T00:00:00Z') });
+    await a.put(ref('item'), { who: 'a' }, { parentVersion: null, actor: 't' });
+    await b.put(ref('item'), { who: 'b' }, { parentVersion: null, actor: 't' });
+    const layered = new LayeredRepository({
+      layers: [{ label: 'a', repo: a }, { label: 'b', repo: b }],
+    });
+    const events: MetadataEvent[] = [];
+    for await (const evt of layered.history(ref('item'))) events.push(evt);
+    expect(events).toHaveLength(2);
+    expect(events[0]!.source.startsWith('a:')).toBe(true);
+    expect(events[1]!.source.startsWith('b:')).toBe(true);
+  });
+});