fix(core): sync credentials across locations

packages/core/src/bus.ts

@@ -89,6 +89,8 @@ export interface PublishOptions {
   readonly id?: Event.ID
   readonly metadata?: Record<string, unknown>
   readonly location?: Location.Ref
+  /** Publishes without Location metadata so every Location-scoped subscriber receives the event. */
+  readonly global?: boolean
   /** Local operational projection committed atomically with a new durable event. Not replayed or serialized. */
   readonly commit?: (seq: number) => Effect.Effect<void>
 }
@@ -450,11 +452,12 @@ 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 =
+            const location = options?.global
-              options?.location ??
+              ? undefined
-              (serviceLocation
+              : (options?.location ??
-                ? { directory: serviceLocation.directory, workspaceID: serviceLocation.workspaceID }
+                (serviceLocation
-                : undefined)
+                  ? { directory: serviceLocation.directory, workspaceID: serviceLocation.workspaceID }
+                  : undefined))
             return yield* publishEvent(
               definition,
               {
@@ -484,11 +487,12 @@ export function configured(options?: Options) {
                     }),
                   )
                 }
-                const location =
+                const location = options?.global
-                  options?.location ??
+                  ? undefined
-                  (serviceLocation
+                  : (options?.location ??
-                    ? { directory: serviceLocation.directory, workspaceID: serviceLocation.workspaceID }
+                    (serviceLocation
-                    : undefined)
+                      ? { directory: serviceLocation.directory, workspaceID: serviceLocation.workspaceID }
+                      : undefined))
                 return {
                   definition,
                   aggregateID,

packages/core/src/integration.ts

@@ -421,7 +421,11 @@ const layer = Layer.effect(
             // Persisting attempts cannot be cancelled, expired, or claimed again.
             yield* SynchronizedRef.update(attempts, (current) => new Map(current).set(attemptID, terminal))
             if (Exit.isFailure(persistence)) yield* Effect.failCause(persistence.cause)
-            yield* bus.publish(Integration.Event.ConnectionUpdated, { integrationID: attempt.integrationID })
+            yield* bus.publish(
+              Integration.Event.ConnectionUpdated,
+              { integrationID: attempt.integrationID },
+              { global: true },
+            )
             yield* bus.publish(Integration.Event.Updated, {})
           }).pipe(Effect.ensuring(close(attempt.scope)))
         }),
@@ -477,7 +481,11 @@ const layer = Layer.effect(
           yield* SynchronizedRef.update(commandAttempts, (current) => new Map(current).set(attemptID, terminal))
           yield* close(attempt.scope)
           if (Exit.isFailure(persistence)) return
-          yield* bus.publish(Integration.Event.ConnectionUpdated, { integrationID: attempt.integrationID })
+          yield* bus.publish(
+            Integration.Event.ConnectionUpdated,
+            { integrationID: attempt.integrationID },
+            { global: true },
+          )
           yield* bus.publish(Integration.Event.Updated, {})
         }),
       )
@@ -686,6 +694,11 @@ const layer = Layer.effect(
           if (credential.value.expires > now + Duration.toMillis(Duration.minutes(5))) return credential.value
           const value = yield* authorize(implementation.refresh(credential.value))
           yield* credentials.update(credential.id, { value })
+          yield* bus.publish(
+            Integration.Event.ConnectionUpdated,
+            { integrationID: credential.integrationID },
+            { global: true },
+          )
           return value
         }),
         key: Effect.fn("Integration.connection.key")(function* (input) {
@@ -711,14 +724,22 @@ const layer = Layer.effect(
               ...(Object.keys(answer).length > 0 ? { configuration: answer } : {}),
             }),
           })
-          yield* bus.publish(Integration.Event.ConnectionUpdated, { integrationID: input.integrationID })
+          yield* bus.publish(
+            Integration.Event.ConnectionUpdated,
+            { integrationID: input.integrationID },
+            { global: true },
+          )
           yield* bus.publish(Integration.Event.Updated, {})
         }),
         update: Effect.fn("Integration.connection.update")(function* (credentialID, updates) {
           const credential = yield* credentials.get(credentialID)
           yield* credentials.update(credentialID, updates)
           if (credential) {
-            yield* bus.publish(Integration.Event.ConnectionUpdated, { integrationID: credential.integrationID })
+            yield* bus.publish(
+              Integration.Event.ConnectionUpdated,
+              { integrationID: credential.integrationID },
+              { global: true },
+            )
           }
           yield* bus.publish(Integration.Event.Updated, {})
         }),
@@ -726,7 +747,11 @@ const layer = Layer.effect(
           const credential = yield* credentials.get(credentialID)
           yield* credentials.remove(credentialID)
           if (credential) {
-            yield* bus.publish(Integration.Event.ConnectionUpdated, { integrationID: credential.integrationID })
+            yield* bus.publish(
+              Integration.Event.ConnectionUpdated,
+              { integrationID: credential.integrationID },
+              { global: true },
+            )
           }
           yield* bus.publish(Integration.Event.Updated, {})
         }),

packages/core/test/location-layer.test.ts

@@ -486,6 +486,45 @@ describe("LocationServiceMap", () => {
     ),
   )
 
+  it.live("routes global events to every location", () =>
+    Effect.acquireRelease(
+      Effect.promise(() => Promise.all([tmpdir(), tmpdir()])),
+      (dirs) => Effect.promise(() => Promise.all(dirs.map((dir) => dir[Symbol.asyncDispose]())).then(() => undefined)),
+    ).pipe(
+      Effect.flatMap(([first, second]) =>
+        Effect.scoped(
+          Effect.gen(function* () {
+            const locations = yield* LocationServiceMap.Service
+            const bus = yield* Bus.Service
+            const firstContext = yield* locations.contextEffect(
+              Location.Ref.make({ directory: AbsolutePath.make(first.path) }),
+            )
+            const secondContext = yield* locations.contextEffect(
+              Location.Ref.make({ directory: AbsolutePath.make(second.path) }),
+            )
+            const received = { first: 0, second: 0 }
+            yield* bus.subscribe(Config.Event.Updated).pipe(
+              Stream.runForEach(() => Effect.sync(() => received.first++)),
+              Effect.provideContext(firstContext),
+              Effect.forkScoped({ startImmediately: true }),
+            )
+            yield* bus.subscribe(Config.Event.Updated).pipe(
+              Stream.runForEach(() => Effect.sync(() => received.second++)),
+              Effect.provideContext(secondContext),
+              Effect.forkScoped({ startImmediately: true }),
+            )
+            yield* Effect.sleep("10 millis")
+
+            yield* bus.publish(Config.Event.Updated, {}, { global: true })
+            yield* Effect.sleep("10 millis")
+
+            expect(received).toEqual({ first: 1, second: 1 })
+          }),
+        ),
+      ),
+    ),
+  )
+
   it.live("reuses cached services for constructed and decoded location refs", () =>
     Effect.acquireRelease(
       Effect.promise(() => tmpdir()),