fix(core): broadcast connection updates to every location

packages/core/src/bus.ts

@@ -416,11 +416,14 @@ export function configured(options?: Options) {
       function publish<D extends Event.Definition>(definition: D, data: Event.Data<D>, options?: PublishOptions) {
         return Effect.gen(function* () {
           const serviceLocation = Option.getOrUndefined(yield* Effect.serviceOption(Location.Service))
-          const location =
-            options?.location ??
-            (serviceLocation
-              ? { directory: serviceLocation.directory, workspaceID: serviceLocation.workspaceID }
-              : undefined)
+          // Global definitions describe location-independent facts. Never tag
+          // them, so location-filtered subscribers in every location observe them.
+          const location = definition.global
+            ? undefined
+            : (options?.location ??
+              (serviceLocation
+                ? { directory: serviceLocation.directory, workspaceID: serviceLocation.workspaceID }
+                : undefined))
           return yield* publishEvent(
             definition,
             {

packages/core/test/bus.test.ts

@@ -75,6 +75,13 @@ const CountMessage = Bus.ephemeral({
     count: Schema.Number,
   },
 })
+const GlobalFact = Bus.ephemeral({
+  type: "test.global.fact",
+  global: true,
+  schema: {
+    text: Schema.String,
+  },
+})
 
 const VersionedMessage = Bus.durable({
   type: "test.versioned",
@@ -153,6 +160,31 @@ describe("Bus", () => {
     }),
   )
 
+  it.effect("publishes global definitions untagged so subscribers in other locations observe them", () =>
+    Effect.gen(function* () {
+      const bus = yield* Bus.Service
+      const elsewhere = Location.Service.of(
+        location({ directory: AbsolutePath.make("elsewhere"), workspaceID: Workspace.ID.make("wrk_other") }),
+      )
+      const fiber = yield* bus
+        .subscribe([Message, GlobalFact])
+        .pipe(
+          Stream.take(1),
+          Stream.runCollect,
+          Effect.provideService(Location.Service, elsewhere),
+          Effect.forkScoped,
+        )
+      yield* Effect.yieldNow
+
+      // Location-tagged events stay invisible to other locations; the global fact reaches them.
+      yield* bus.publish(Message, { text: "tagged" })
+      const event = yield* bus.publish(GlobalFact, { text: "everywhere" })
+
+      expect(event).not.toHaveProperty("location")
+      expect(Array.from(yield* Fiber.join(fiber))).toEqual([event])
+    }),
+  )
+
   itWithoutLocation.effect("omits location when no location is available", () =>
     Effect.gen(function* () {
       const bus = yield* Bus.Service

packages/schema/src/event.ts

@@ -37,6 +37,7 @@ export type DurableDefinition<
     readonly version: number
     readonly aggregate: string
   }
+  readonly global?: never
   readonly data: DataSchema
 }
 
@@ -47,6 +48,8 @@ export type EphemeralDefinition<
   readonly type: Type
   readonly durability: "ephemeral"
   readonly durable?: never
+  /** Global events describe location-independent facts: they are published untagged and reach every location. */
+  readonly global?: boolean
   readonly data: DataSchema
 }
 
@@ -77,13 +80,14 @@ type Input<Type extends string, Fields extends Readonly<Record<PropertyKey, Sche
     readonly version: number
     readonly aggregate: string
   }
+  readonly global?: boolean
   readonly schema: Fields
 }
 
 export function durable<
   const Type extends string,
   const Fields extends Readonly<Record<PropertyKey, Schema.Codec<unknown, unknown>>>,
->(input: Input<Type, Fields> & { readonly durable: NonNullable<Input<Type, Fields>["durable"]> }) {
+>(input: Omit<Input<Type, Fields>, "global"> & { readonly durable: NonNullable<Input<Type, Fields>["durable"]> }) {
   const data = Schema.Struct(input.schema)
   const durable = Schema.Struct({
     aggregateID: DurableEnvelope.fields.aggregateID,
@@ -137,6 +141,7 @@ export function ephemeral<
         type: input.type,
         durability: "ephemeral" as const,
         durable: undefined,
+        global: input.global === true,
         data,
       })),
     ) satisfies EphemeralDefinition<Type, typeof data>

packages/schema/src/integration.ts

@@ -88,8 +88,12 @@ const Updated = ephemeral({
   type: "integration.updated",
   schema: {},
 })
+// Credentials live in one global store shared by every location, so a
+// connection change is a location-independent fact: publish it globally so
+// every active location refreshes its provider catalog.
 const ConnectionUpdated = ephemeral({
   type: "integration.connection.updated",
+  global: true,
   schema: { integrationID: ID },
 })
 export const Event = { Updated, ConnectionUpdated, Definitions: inventory(Updated, ConnectionUpdated) }