|
| 1 | +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. |
| 2 | + |
| 3 | +/** |
| 4 | + * [#25467 cut 25] The post-persistence mutation fan-out of |
| 5 | + * `ObjectStackProtocolImplementation`, moved out of `protocol.ts` whole: the |
| 6 | + * four members that tell the rest of the process — and, through the cluster |
| 7 | + * transport, the rest of the fleet — that a metadata write has landed. |
| 8 | + * |
| 9 | + * - `runMutationProjector` — the registered projector for a just-persisted |
| 10 | + * mutation (ADR-0094), awaited, never throwing. |
| 11 | + * - `notifyMutationListenersLocal` — the LOCAL half of |
| 12 | + * `emitMetadataMutation` (#2588, #13331): each listener isolated. |
| 13 | + * - `publishMetadataMutation` — the cluster half of `emitMetadataMutation` |
| 14 | + * (#13331): the mutation's address to peers, fire-and-forget, drafts |
| 15 | + * skipped. |
| 16 | + * - `emitMetaItemPublished` — the per-item publish listeners (#10219), |
| 17 | + * awaited and isolated. |
| 18 | + * |
| 19 | + * A MOVE, not a redesign. Each function below is the body of the |
| 20 | + * `ObjectStackProtocolImplementation` method of the same name as it stood, |
| 21 | + * re-indented one level and otherwise unedited: the same order, the same |
| 22 | + * isolation, the same log lines. |
| 23 | + * |
| 24 | + * ## How the moved bodies reach the class |
| 25 | + * |
| 26 | + * The shape #23050 set for `rest-server.ts` (its `data-action-routes.ts` |
| 27 | + * header carries the full reasoning), as `protocol-batch-data.ts` applies it. |
| 28 | + * The class keeps each method — the published `dist/*.d.ts` lists its name, |
| 29 | + * docblock and declared signature — and its body is one literal of exactly |
| 30 | + * the {@link MutationFanOutHost} members it reads, plus a `.call`. Every |
| 31 | + * member read is a field, so each is handed over as a getter: the body reads |
| 32 | + * the protocol's own listener arrays, projector map and cluster wiring at |
| 33 | + * call time — `attachMetadataMutationPubSub` and |
| 34 | + * `detachMetadataMutationPubSub` reassign `clusterPubSub` and |
| 35 | + * `clusterNodeId`, and the getters see the current values, as the bodies did. |
| 36 | + * |
| 37 | + * The callers stay in the class and keep calling the class methods: |
| 38 | + * `emitMetadataMutation` (local, then cluster), `applyRemoteMetadataMutation` |
| 39 | + * (local only), the seven projecting doors and the per-item publish door. So |
| 40 | + * the call-site counts that `protocol.recovery-doors-emit-mutation.test.ts` |
| 41 | + * and `protocol.recovery-doors-mutation-projector.test.ts` take over |
| 42 | + * `protocol.ts` do not move. |
| 43 | + * |
| 44 | + * What the bodies read from elsewhere, they import from where `protocol.ts` |
| 45 | + * declares it. Nothing re-exports this module, so the published surface does |
| 46 | + * not move. |
| 47 | + */ |
| 48 | + |
| 49 | +import type { |
| 50 | + ClusterMetadataMutationPayload, |
| 51 | + MetadataMutationEvent, |
| 52 | + MetaItemPublishedEvent, |
| 53 | + MutationProjectionOutcome, |
| 54 | + ObjectStackProtocolImplementation, |
| 55 | +} from './protocol.js'; |
| 56 | +import { METADATA_MUTATION_CLUSTER_CHANNEL } from './protocol.js'; |
| 57 | + |
| 58 | +/** |
| 59 | + * The `ObjectStackProtocolImplementation` members the functions below read — |
| 60 | + * every one of them, and no other; each function's `this` is the `Pick` of |
| 61 | + * the members it reads. Each is typed AS the class member (an indexed access |
| 62 | + * on the class), so a change to a member's type on the class is a change |
| 63 | + * here too, and naming a member the class does not have is a compile error. |
| 64 | + */ |
| 65 | +export interface MutationFanOutHost { |
| 66 | + readonly mutationProjectors: ObjectStackProtocolImplementation['mutationProjectors']; |
| 67 | + readonly metadataMutationListeners: ObjectStackProtocolImplementation['metadataMutationListeners']; |
| 68 | + readonly clusterPubSub: ObjectStackProtocolImplementation['clusterPubSub']; |
| 69 | + readonly clusterNodeId: ObjectStackProtocolImplementation['clusterNodeId']; |
| 70 | + readonly metaItemPublishedListeners: ObjectStackProtocolImplementation['metaItemPublishedListeners']; |
| 71 | +} |
| 72 | + |
| 73 | +/** The body of `ObjectStackProtocolImplementation#runMutationProjector`; the method's docblock there is its contract. */ |
| 74 | +export async function runMutationProjector( |
| 75 | + this: Pick<MutationFanOutHost, 'mutationProjectors'>, |
| 76 | + evt: MetadataMutationEvent & { body?: unknown }, |
| 77 | +): Promise<MutationProjectionOutcome | undefined> { |
| 78 | + const projector = this.mutationProjectors.get(evt.type); |
| 79 | + if (!projector) return undefined; |
| 80 | + try { |
| 81 | + await projector(evt); |
| 82 | + return { success: true }; |
| 83 | + } catch (e) { |
| 84 | + const error = e instanceof Error ? e.message : String(e); |
| 85 | + console.warn( |
| 86 | + `[Protocol] mutation projector failed for ${evt.type}/${evt.name} (state=${evt.state}): ${error}`, |
| 87 | + ); |
| 88 | + return { success: false, error }; |
| 89 | + } |
| 90 | +} |
| 91 | + |
| 92 | +/** The body of `ObjectStackProtocolImplementation#notifyMutationListenersLocal`; the method's docblock there is its contract. */ |
| 93 | +export function notifyMutationListenersLocal( |
| 94 | + this: Pick<MutationFanOutHost, 'metadataMutationListeners'>, |
| 95 | + evt: MetadataMutationEvent, |
| 96 | +): void { |
| 97 | + for (const listener of this.metadataMutationListeners) { |
| 98 | + try { |
| 99 | + listener(evt); |
| 100 | + } catch (e) { |
| 101 | + console.warn( |
| 102 | + `[Protocol] metadata-mutation listener failed for ${evt.type}/${evt.name}: ` |
| 103 | + + `${e instanceof Error ? e.message : String(e)}`, |
| 104 | + ); |
| 105 | + } |
| 106 | + } |
| 107 | +} |
| 108 | + |
| 109 | +/** The body of `ObjectStackProtocolImplementation#publishMetadataMutation`; the method's docblock there is its contract. */ |
| 110 | +export function publishMetadataMutation( |
| 111 | + this: Pick<MutationFanOutHost, 'clusterPubSub' | 'clusterNodeId'>, |
| 112 | + evt: MetadataMutationEvent, |
| 113 | +): void { |
| 114 | + if (!this.clusterPubSub) return; |
| 115 | + if (evt.state === 'draft') return; |
| 116 | + const payload: ClusterMetadataMutationPayload = { |
| 117 | + originNode: this.clusterNodeId, |
| 118 | + event: evt, |
| 119 | + }; |
| 120 | + void this.clusterPubSub |
| 121 | + .publish(METADATA_MUTATION_CLUSTER_CHANNEL, payload, { |
| 122 | + partitionKey: `${evt.type}:${evt.name}`, |
| 123 | + }) |
| 124 | + .catch((err) => { |
| 125 | + console.warn( |
| 126 | + `[Protocol] cluster metadata-mutation publish failed for ` |
| 127 | + + `${evt.type}/${evt.name}: ${err instanceof Error ? err.message : String(err)}`, |
| 128 | + ); |
| 129 | + }); |
| 130 | +} |
| 131 | + |
| 132 | +/** The body of `ObjectStackProtocolImplementation#emitMetaItemPublished`; the method's docblock there is its contract. */ |
| 133 | +export async function emitMetaItemPublished( |
| 134 | + this: Pick<MutationFanOutHost, 'metaItemPublishedListeners'>, |
| 135 | + evt: MetaItemPublishedEvent, |
| 136 | +): Promise<void> { |
| 137 | + for (const listener of this.metaItemPublishedListeners) { |
| 138 | + try { |
| 139 | + await listener(evt); |
| 140 | + } catch (e) { |
| 141 | + console.warn( |
| 142 | + `[Protocol] the post-publish re-bind announce FAILED for ${evt.type}/${evt.name} — the item IS ` |
| 143 | + + `published and stored, but boot-cached consumers keep the PRE-publish view until this process ` |
| 144 | + + `restarts: a newly published record-triggered flow does not bind its trigger (it will not fire), ` |
| 145 | + + `an edited schedule-triggered flow keeps running its old definition, and authored hooks, actions ` |
| 146 | + + `and translations are not re-synced. Nothing retries this announce. Re-publish the item once the ` |
| 147 | + + `cause below is resolved (it is idempotent), or restart the process to rebuild every subscriber ` |
| 148 | + + `from storage. Cause: ${e instanceof Error ? e.message : String(e)}`, |
| 149 | + ); |
| 150 | + } |
| 151 | + } |
| 152 | +} |
0 commit comments