web-terminal/control-plane/src/routing/bus.ts

/**
 * RevocationBus implementations over the FROZEN `relay:revocations` channel (INDEX §4.2 FIX 4).
 * P3 only PUBLISHES (P1 owns the subscriber that injects §4.1 CLOSE+RST / GOAWAY). Drain (T10)
 * and revoke (T13) use the SAME named channel. The KillSignal shape + channel name are imported
 * from relay-contracts and NEVER redefined.
 */
import {
  RELAY_REVOCATIONS_CHANNEL,
  KillSignalSchema,
  type KillSignal,
  type RevocationBus,
} from 'relay-contracts'

/** In-memory bus with a subscribe hook — used by tests and single-process wiring. */
export interface TestableRevocationBus extends RevocationBus {
  subscribe(handler: (signal: KillSignal) => void): () => void
  readonly channel: typeof RELAY_REVOCATIONS_CHANNEL
}

export function createInMemoryRevocationBus(): TestableRevocationBus {
  const handlers = new Set<(signal: KillSignal) => void>()
  return {
    channel: RELAY_REVOCATIONS_CHANNEL,
    async publish(signal) {
      // Validate at the boundary before it hits the wire (INV10: reason is metadata only).
      const parsed = KillSignalSchema.parse(signal)
      for (const h of handlers) h(parsed as KillSignal)
    },
    subscribe(handler) {
      handlers.add(handler)
      return () => handlers.delete(handler)
    },
  }
}

/** Minimal publisher surface of an ioredis client (integration seam; typed, not unit-tested). */
export interface RedisPublisher {
  publish(channel: string, message: string): Promise<number>
}

/** Redis-backed publisher (production). P1 relay nodes SUBSCRIBE to the same channel. */
export function createRedisRevocationBus(redis: RedisPublisher): RevocationBus {
  return {
    async publish(signal) {
      const parsed = KillSignalSchema.parse(signal)
      await redis.publish(RELAY_REVOCATIONS_CHANNEL, JSON.stringify(parsed))
    },
  }
}