fix(metadata): register()/unregister() announce to subscribe() watchers (#3112) (#3131)

* fix(metadata): register()/unregister() announce to subscribe() watchers (#3112)

`MetadataManager.register()` updated the in-memory registry, persisted to
writable loaders and published to the realtime service — but never called
`notifyWatchers()`. `unregister()` had the same gap. Watchers only ever fired
from the `saveMetaItem` path (indirectly, via the sys_metadata repository watch
stream) and the filesystem watcher, so `subscribe()` looked like it covered
every write while silently missing all of them.

The consumer that pays for this is ObjectQL's SchemaRegistry bridge — the
component that decides what is queryable. #3108 worked around it for the
`metadata:reloaded` seam by carrying the parsed artifact on the event payload,
but the asymmetry remained for every other `register()` call site and every
other `subscribe()` consumer.

Announce by default, so a new call site is correct without knowing the contract
exists. Two concrete fixes fall out: `unregisterPackage()` now tells cached
consumers to drop uninstalled objects instead of serving them until restart,
and runtime datasource writes reach watchers.

Bulk ingest opts out explicitly via the new `MetadataWriteOptions`
(`{ notify: false }`) at the five boot/batch sites, each of which either runs
before consumers cache anything or announces the whole batch once. ObjectQL's
registry bridge MUST stay silent for a second reason: it copies objects OUT of
the SchemaRegistry, and announcing would feed them back through a handler that
re-registers under `_packageId ?? 'metadata-service'` — overwriting the true
package provenance of every object whose body carries no `_packageId`.

Additive only: the 3-arg forms keep working, and the api-surface ratchet reports
0 breaking / 1 added.

Tests pin both halves of the contract (announce by default, silence only when
asked) and the no-flood invariant on the reload path; both were verified to fail
against the pre-change behavior.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* docs(changeset): state the register() notify change as a contract fix, not a bug fix

unregisterPackage()/bulkUnregister() have no production caller in framework or
cloud today, so their announce is correct-but-latent. The one live behavior
change is runtime datasource writes reaching the HMR SSE stream. Say so instead
of advertising two 'concrete fixes'.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: Jack Zhuang <277994282+os-zhuang@users.noreply.github.com>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: os-zhuang

.changeset/metadata-register-notifies-watchers.md

@@ -0,0 +1,15 @@
+---
+"@objectstack/spec": patch
+"@objectstack/metadata": patch
+"@objectstack/objectql": patch
+---
+
+**`MetadataManager.register()` / `unregister()` now announce to `subscribe()` watchers.** Both updated the registry, persisted to writable loaders and published to realtime, but never fired the watch callbacks — so `subscribe()` looked like it covered every write while silently missing all of them. Only the `saveMetaItem` path (via the repository watch stream) and the filesystem watcher ever reached a subscriber. Runtime consumers that cache metadata — notably ObjectQL's SchemaRegistry bridge, the component that decides what is queryable — went stale on every other write until the process restarted.
+
+Announcing is now the **default**, so a new call site is correct without knowing this contract exists. This is a contract fix rather than a bug fix: the one live behavior change is that runtime datasource writes (`datasource-admin`) now reach the HMR SSE stream, which subscribes to every registered type. `unregisterPackage()` / `bulkUnregister()` also announce their deletes now — correct, but latent, since neither has a production caller today.
+
+Bulk ingest opts out explicitly with the new `MetadataWriteOptions` (`{ notify: false }`) — boot-time filesystem priming, artifact ingest, and ObjectQL's registry bridge, each of which either runs before consumers cache anything or announces the whole batch once (as the artifact reload path does via `metadata:reloaded`). The bridge in particular MUST stay silent: it copies objects out of the SchemaRegistry, and announcing would feed them back through a handler that re-registers under `_packageId ?? 'metadata-service'`, overwriting the true package provenance of every object whose body carries no `_packageId`.
+
+Additive only — `register(type, name, data)` and `unregister(type, name)` keep working unchanged.
+
+Fixes #3112.

packages/metadata/src/metadata-manager.ts

@@ -28,6 +28,7 @@ import type {
   MetadataImportOptions,
   MetadataImportResult,
   MetadataTypeInfo,
+  MetadataWriteOptions,
   IRealtimeService,
   RealtimeEventPayload,
   IPubSub,
@@ -276,8 +277,20 @@ export class MetadataManager implements IMetadataService {
    * Stores in-memory registry and persists to database-backed loaders only.
    * FilesystemLoader (protocol 'file:') is read-only for static metadata and
    * should not be written to during runtime registration.
+   *
+   * Announces the write to {@link subscribe} watchers as an `added` /
+   * `changed` {@link MetadataWatchEvent}, so consumers that cache metadata
+   * (ObjectQL's SchemaRegistry bridge, the HMR SSE stream) refresh instead of
+   * serving the pre-write definition until restart. Pass `{ notify: false }`
+   * for bulk ingest that announces by other means — read
+   * {@link MetadataWriteOptions.notify} before doing so.
    */
-  async register(type: string, name: string, data: unknown): Promise<void> {
+  async register(
+    type: string,
+    name: string,
+    data: unknown,
+    options?: MetadataWriteOptions,
+  ): Promise<void> {
     // Persistence write gate: when `persistence.writable` is explicitly false
     // we treat register() as read-only. Default `true` (or omitted) preserves
     // historical behavior.
@@ -290,6 +303,11 @@ export class MetadataManager implements IMetadataService {
       return;
     }
 
+    // Captured before the write so the event distinguishes a first
+    // registration from an overwrite, matching the repo-watch path's
+    // 'added' vs 'changed' split.
+    const existed = this.registry.get(type)?.has(name) ?? false;
+
     if (!this.registry.has(type)) {
       this.registry.set(type, new Map());
     }
@@ -326,6 +344,20 @@ export class MetadataManager implements IMetadataService {
         this.logger.warn(`Failed to publish metadata event`, { type, name, error });
       }
     }
+
+    // Announce last, once the write has landed in the registry and every
+    // writable loader — a subscriber that re-reads on the event must not
+    // race ahead of the data it is meant to observe.
+    if (options?.notify !== false) {
+      this.notifyWatchers(type, {
+        type: existed ? 'changed' : 'added',
+        metadataType: type,
+        name,
+        path: '',
+        data,
+        timestamp: new Date().toISOString(),
+      });
+    }
   }
 
   /**
@@ -336,6 +368,13 @@ export class MetadataManager implements IMetadataService {
    * Addendum) declared in `*.datasource.ts` and owned by source control. Writing
    * them through `register()` would persist them to `sys_metadata` and create
    * drift between the artefact and the DB; this method avoids that.
+   *
+   * Deliberately silent: it does NOT announce to {@link subscribe} watchers.
+   * This is a boot-time seeding primitive for artefacts that source control
+   * owns — callers that mutate metadata mid-run want {@link register}, which
+   * announces. If you add a mid-run caller here, announce the change yourself
+   * (as the artifact reload path does via `metadata:reloaded`) or its
+   * consumers will read the pre-write definition until restart.
    */
   registerInMemory(type: string, name: string, data: unknown): void {
     if (!this.registry.has(type)) {
@@ -415,8 +454,13 @@ export class MetadataManager implements IMetadataService {
   /**
    * Unregister/remove a metadata item by type and name.
    * Deletes from database-backed loaders only (same rationale as register()).
+   *
+   * Announces the removal to {@link subscribe} watchers as a `deleted`
+   * {@link MetadataWatchEvent} — the delete half of the {@link register}
+   * contract. Pass `{ notify: false }` only for teardown that announces by
+   * other means.
    */
-  async unregister(type: string, name: string): Promise<void> {
+  async unregister(type: string, name: string, options?: MetadataWriteOptions): Promise<void> {
     // Remove from in-memory registry
     const typeStore = this.registry.get(type);
     if (typeStore) {
@@ -458,6 +502,18 @@ export class MetadataManager implements IMetadataService {
         this.logger.warn(`Failed to publish metadata event`, { type, name, error });
       }
     }
+
+    // Announce last, once the removal has landed everywhere (see register()).
+    if (options?.notify !== false) {
+      this.notifyWatchers(type, {
+        type: 'deleted',
+        metadataType: type,
+        name,
+        path: '',
+        data: undefined,
+        timestamp: new Date().toISOString(),
+      });
+    }
   }
 
   /**
@@ -892,20 +948,24 @@ export class MetadataManager implements IMetadataService {
   // ==========================================
 
   /**
-   * Register multiple metadata items in a single batch
+   * Register multiple metadata items in a single batch.
+   *
+   * Announces one event per item, like {@link register}. Pass
+   * `{ notify: false }` when the batch is boot-time ingest or when the caller
+   * announces the whole set once — see {@link MetadataWriteOptions.notify}.
    */
   async bulkRegister(
     items: Array<{ type: string; name: string; data: unknown }>,
-    options?: { continueOnError?: boolean; validate?: boolean }
+    options?: { continueOnError?: boolean; validate?: boolean } & MetadataWriteOptions
   ): Promise<MetadataBulkResult> {
-    const { continueOnError = false } = options ?? {};
+    const { continueOnError = false, notify } = options ?? {};
     let succeeded = 0;
     let failed = 0;
     const errors: Array<{ type: string; name: string; error: string }> = [];
 
     for (const item of items) {
       try {
-        await this.register(item.type, item.name, item.data);
+        await this.register(item.type, item.name, item.data, { notify });
         succeeded++;
       } catch (e) {
         failed++;
@@ -927,16 +987,21 @@ export class MetadataManager implements IMetadataService {
   }
 
   /**
-   * Unregister multiple metadata items in a single batch
+   * Unregister multiple metadata items in a single batch.
+   *
+   * Announces one `deleted` event per item, like {@link unregister}.
    */
-  async bulkUnregister(items: Array<{ type: string; name: string }>): Promise<MetadataBulkResult> {
+  async bulkUnregister(
+    items: Array<{ type: string; name: string }>,
+    options?: MetadataWriteOptions,
+  ): Promise<MetadataBulkResult> {
     let succeeded = 0;
     let failed = 0;
     const errors: Array<{ type: string; name: string; error: string }> = [];
 
     for (const item of items) {
       try {
-        await this.unregister(item.type, item.name);
+        await this.unregister(item.type, item.name, options);
         succeeded++;
       } catch (e) {
         failed++;

packages/metadata/src/plugin-hmr-reload.test.ts

@@ -90,6 +90,34 @@ describe('MetadataPlugin._reloadAndAnnounce — fires metadata:reloaded after re
         ]);
     });
 
+    it('announces ONCE per reload, not once per ingested item (#3112)', async () => {
+        // `register()` announces to subscribe() watchers by default. Artifact
+        // ingest deliberately opts out (`{ notify: false }`) because this hook
+        // is the announcement for the whole batch. If someone drops that
+        // opt-out, every reload fans N per-item events into every watcher —
+        // each one racing the ingest that is still landing — and boot-time
+        // registration floods subscribers that have nothing to refresh yet.
+        const plugin = new MetadataPlugin({
+            watch: false,
+            config: { bootstrap: 'eager' },
+            environmentId: 'proj_test',
+        });
+        const mgr = (plugin as any).manager as NodeMetadataManager;
+        const ctx = fakeCtx();
+        const file = writeArtifact('sweep3');
+
+        const perItemEvents: unknown[] = [];
+        mgr.subscribe('flow', (evt) => perItemEvents.push(evt));
+
+        await (plugin as any)._reloadAndAnnounce(ctx, { path: file, fetchTimeoutMs: undefined }, [file]);
+
+        expect(perItemEvents).toEqual([]);
+        // The single batch-level announcement is the contract, and it carries
+        // the bodies consumers need to re-ingest.
+        expect(ctx.trigger).toHaveBeenCalledTimes(1);
+        expect(await mgr.get('flow', 'sweep3')).toBeDefined();
+    });
+
     it('still announces even if a subscriber throws (reload must not break)', async () => {
         const plugin = new MetadataPlugin({
             watch: false,

packages/metadata/src/plugin.ts

@@ -510,6 +510,14 @@ export class MetadataPlugin implements Plugin {
     /**
      * Parse raw artifact JSON (envelope or bare definition) and register all
      * metadata items into the MetadataManager.
+     *
+     * Registers with `{ notify: false }` — one announcement per artifact, not
+     * one per item. Both callers cover the whole set already: the boot load
+     * runs before consumers have cached anything, and the reload path
+     * (`_reloadAndAnnounce`) fires `metadata:reloaded` carrying the parsed
+     * artifact once the ingest is complete. Announcing here too would emit N
+     * duplicate events per reload, each racing the batch that is still
+     * landing.
      */
     private async _parseAndRegisterArtifact(ctx: PluginContext, raw: unknown, label: string): Promise<number> {
         const { EnvironmentArtifactSchema } = await import('@objectstack/spec/cloud');
@@ -568,7 +576,7 @@ export class MetadataPlugin implements Plugin {
                         packageVersion: manifestVersion,
                     });
                     await memLoader.save('view', viewObject, item);
-                    await this.manager.register('view', viewObject, item);
+                    await this.manager.register('view', viewObject, item, { notify: false });
                     totalRegistered++;
                     for (const vi of expandViewContainer(viewObject, item)) {
                         for (const w of vi._diagnostics?.warnings ?? []) {
@@ -579,7 +587,7 @@ export class MetadataPlugin implements Plugin {
                             packageVersion: manifestVersion,
                         });
                         await memLoader.save('view', vi.name, vi);
-                        await this.manager.register('view', vi.name, vi);
+                        await this.manager.register('view', vi.name, vi, { notify: false });
                         totalRegistered++;
                     }
                     continue;
@@ -614,7 +622,7 @@ export class MetadataPlugin implements Plugin {
                     packageVersion: manifestVersion,
                 });
                 await memLoader.save(metaType, name, item);
-                await this.manager.register(metaType, name, item);
+                await this.manager.register(metaType, name, item, { notify: false });
                 totalRegistered++;
             }
         }
@@ -738,7 +746,11 @@ export class MetadataPlugin implements Plugin {
                             applyProtection(meta, {
                                 packageId: this.options.packageId,
                             });
-                            await this.manager.register(entry.type, meta.name, item);
+                            // Silent: boot-time priming, before any consumer
+                            // has cached a definition to go stale. Post-boot
+                            // edits to these files reach watchers through the
+                            // FileSystemRepository attached by onEnable.
+                            await this.manager.register(entry.type, meta.name, item, { notify: false });
                         }
                     }
                     ctx.logger.info(`Loaded ${items.length} ${entry.type} from file system`);