From fee72573bac8704840d03ef496f73857948a6bdb Mon Sep 17 00:00:00 2001 From: Jeff Emmett Date: Wed, 11 Mar 2026 16:20:04 -0700 Subject: [PATCH] feat: system clock heartbeat + ephemeral clock delivery (TASK-47) Server-side SystemClock broadcasts tick/5min/hourly/daily events via WebSocket to all connected clients. Events are ephemeral (not persisted in CRDT). Client-side fallback clock activates when server connection is lost. Co-Authored-By: Claude Opus 4.6 --- ...ck-Heartbeat-Service-for-rSpace-canvas.md} | 31 ++-- lib/community-sync.ts | 7 + lib/event-bus.ts | 86 +++++++++- server/clock-service.ts | 147 ++++++++++++++++++ server/index.ts | 23 +++ 5 files changed, 283 insertions(+), 11 deletions(-) rename backlog/tasks/{task-47 - Implement-System-Clock-heartbeat-service-for-rSpace-canvas.md => task-47 - Implement-System-Clock-Heartbeat-Service-for-rSpace-canvas.md} (71%) create mode 100644 server/clock-service.ts diff --git a/backlog/tasks/task-47 - Implement-System-Clock-heartbeat-service-for-rSpace-canvas.md b/backlog/tasks/task-47 - Implement-System-Clock-Heartbeat-Service-for-rSpace-canvas.md similarity index 71% rename from backlog/tasks/task-47 - Implement-System-Clock-heartbeat-service-for-rSpace-canvas.md rename to backlog/tasks/task-47 - Implement-System-Clock-Heartbeat-Service-for-rSpace-canvas.md index f1c6324..1a051f8 100644 --- a/backlog/tasks/task-47 - Implement-System-Clock-heartbeat-service-for-rSpace-canvas.md +++ b/backlog/tasks/task-47 - Implement-System-Clock-Heartbeat-Service-for-rSpace-canvas.md @@ -1,9 +1,10 @@ --- id: TASK-47 -title: 'Implement System Clock / Heartbeat Service for rSpace canvas' -status: To Do +title: Implement System Clock / Heartbeat Service for rSpace canvas +status: Done assignee: [] created_date: '2026-02-18 22:30' +updated_date: '2026-03-11 23:19' labels: - feature - infrastructure @@ -12,7 +13,9 @@ milestone: m-1 dependencies: - TASK-43 references: - - rspace-online/backlog/tasks/task-43 - Implement-Event-Broadcasting-canvas-wide-pub-sub-system.md + - >- + rspace-online/backlog/tasks/task-43 - + Implement-Event-Broadcasting-canvas-wide-pub-sub-system.md - rSpace-website/docs/R-ECOSYSTEM-ARCHITECTURE.md priority: high --- @@ -81,11 +84,19 @@ Server-level config in community settings: ## Acceptance Criteria -- [ ] #1 SystemClock emits `clock:tick` every 60s via CanvasEventBus -- [ ] #2 Configurable intervals: tick, 5-min, hourly, daily -- [ ] #3 Server-authoritative — only one clock source per canvas -- [ ] #4 Shapes can subscribe to clock channels and receive time payloads -- [ ] #5 Clock events are ephemeral (not persisted in CRDT eventLog ring buffer) -- [ ] #6 Fallback local clock when server connection is lost -- [ ] #7 Clock can be enabled/disabled per community in settings +- [x] #1 SystemClock emits `clock:tick` every 60s via CanvasEventBus +- [x] #2 Configurable intervals: tick, 5-min, hourly, daily +- [x] #3 Server-authoritative — only one clock source per canvas +- [x] #4 Shapes can subscribe to clock channels and receive time payloads +- [x] #5 Clock events are ephemeral (not persisted in CRDT eventLog ring buffer) +- [x] #6 Fallback local clock when server connection is lost +- [x] #7 Clock can be enabled/disabled per community in settings + +## Implementation Notes + + +Implementation started + +Complete. Created server/clock-service.ts with SystemClock class (configurable tick/5min/hourly/daily intervals, min 10s). Wired broadcastClockEvent() in server/index.ts. Updated CommunitySync to handle ephemeral 'clock' WebSocket messages. Updated CanvasEventBus with server clock handling + local fallback (fires when server clock lost for 2.5x interval). Clock events bypass CRDT — WebSocket only. + diff --git a/lib/community-sync.ts b/lib/community-sync.ts index f831c0a..91df18a 100644 --- a/lib/community-sync.ts +++ b/lib/community-sync.ts @@ -367,6 +367,13 @@ export class CommunitySync extends EventTarget { this.dispatchEvent(new CustomEvent("ping-user", { detail: msg })); break; + case "clock": + // Ephemeral clock event from server — dispatch for event bus delivery + this.dispatchEvent(new CustomEvent("clock-event", { + detail: { channel: msg.channel, payload: msg.payload } + })); + break; + case "notification": // Dispatch to window so the notification bell component picks it up window.dispatchEvent(new CustomEvent("rspace-notification", { detail: msg })); diff --git a/lib/event-bus.ts b/lib/event-bus.ts index 2f772b6..c3b886a 100644 --- a/lib/event-bus.ts +++ b/lib/event-bus.ts @@ -6,6 +6,8 @@ * * Events are persisted in the Automerge CRDT doc (bounded ring buffer, * last 100 entries) so remote users see them replayed via sync. + * + * Clock events are ephemeral — delivered via WebSocket but NOT persisted. */ import type { CommunitySync } from "./community-sync"; @@ -29,6 +31,7 @@ export interface EventReceiver { const MAX_LOG_SIZE = 100; const MAX_REENTRY_DEPTH = 10; +const LOCAL_CLOCK_INTERVAL = 60_000; // 60s fallback tick // ── CanvasEventBus ── @@ -36,6 +39,9 @@ export class CanvasEventBus extends EventTarget { #sync: CommunitySync; #processedIds = new Set(); #reentryDepth = 0; + #localClockTimer: ReturnType | null = null; + #serverClockActive = false; + #serverClockTimeout: ReturnType | null = null; constructor(sync: CommunitySync) { super(); @@ -45,6 +51,14 @@ export class CanvasEventBus extends EventTarget { sync.addEventListener("eventlog-changed", () => { this.#processRemoteEvents(); }); + + // Handle ephemeral clock events from server + sync.addEventListener("clock-event", ((e: CustomEvent) => { + this.#handleServerClock(e.detail.channel, e.detail.payload); + }) as EventListener); + + // Start local fallback clock (will be suppressed when server clock is active) + this.#startLocalClock(); } /** @@ -108,6 +122,76 @@ export class CanvasEventBus extends EventTarget { return this.#sync.getShapesSubscribedTo(channel); } + /** Clean up timers. */ + destroy(): void { + this.#stopLocalClock(); + if (this.#serverClockTimeout) { + clearTimeout(this.#serverClockTimeout); + this.#serverClockTimeout = null; + } + } + + // ── Clock handling ── + + /** + * Handle an ephemeral clock event from the server. + * Resets the server-active flag so the local fallback stays suppressed. + */ + #handleServerClock(channel: string, payload: unknown): void { + this.#serverClockActive = true; + + // Reset server-alive timeout (2.5x the tick interval) + if (this.#serverClockTimeout) clearTimeout(this.#serverClockTimeout); + this.#serverClockTimeout = setTimeout(() => { + this.#serverClockActive = false; + console.log("[EventBus] Server clock lost, local fallback active"); + }, LOCAL_CLOCK_INTERVAL * 2.5); + + // Deliver as ephemeral event (no sourceShapeId — it's the system) + this.#deliverEvent({ + id: `clock-${Date.now()}`, + channel, + sourceShapeId: "__system__", + payload, + timestamp: Date.now(), + }); + } + + /** + * Local fallback clock — fires when server clock is not active. + */ + #startLocalClock(): void { + if (this.#localClockTimer) return; + this.#localClockTimer = setInterval(() => { + if (this.#serverClockActive) return; // Server is authoritative + const now = new Date(); + this.#deliverEvent({ + id: `local-clock-${Date.now()}`, + channel: "clock:tick", + sourceShapeId: "__local__", + payload: { + timestamp: now.getTime(), + isoString: now.toISOString(), + hour: now.getUTCHours(), + minute: now.getUTCMinutes(), + second: now.getUTCSeconds(), + dayOfWeek: now.getUTCDay(), + tickCount: -1, // Local fallback marker + }, + timestamp: now.getTime(), + }); + }, LOCAL_CLOCK_INTERVAL); + } + + #stopLocalClock(): void { + if (this.#localClockTimer) { + clearInterval(this.#localClockTimer); + this.#localClockTimer = null; + } + } + + // ── Event processing ── + /** * Process events that arrived via remote Automerge sync. * Skips any events we already delivered locally. @@ -142,7 +226,7 @@ export class CanvasEventBus extends EventTarget { this.#reentryDepth++; try { for (const shapeId of subscribers) { - // Don't deliver back to emitter + // Don't deliver back to emitter (skip for system clock) if (shapeId === entry.sourceShapeId) continue; const shape = this.#sync.getShapeElement(shapeId); diff --git a/server/clock-service.ts b/server/clock-service.ts new file mode 100644 index 0000000..dd83a1a --- /dev/null +++ b/server/clock-service.ts @@ -0,0 +1,147 @@ +/** + * System Clock / Heartbeat Service + * + * Broadcasts consistent time signals to all connected canvas clients. + * Single server-authoritative source — clients subscribe via CanvasEventBus. + * Clock events are ephemeral (WebSocket only, not persisted in CRDT). + */ + +export interface ClockPayload { + timestamp: number; + isoString: string; + hour: number; + minute: number; + second: number; + dayOfWeek: number; + tickCount: number; +} + +export interface ClockConfig { + enabled: boolean; + /** Timezone for daily boundary (default: UTC) */ + timezone: string; + /** Tick interval in seconds (minimum 10, default 60) */ + tickInterval: number; + /** Which channels to emit */ + channels: ("tick" | "minute:5" | "hourly" | "daily")[]; +} + +const DEFAULT_CONFIG: ClockConfig = { + enabled: true, + timezone: "UTC", + tickInterval: 60, + channels: ["tick", "minute:5", "hourly", "daily"], +}; + +export type ClockBroadcastFn = (channel: string, payload: ClockPayload) => void; + +export class SystemClock { + #config: ClockConfig; + #broadcast: ClockBroadcastFn; + #timer: ReturnType | null = null; + #tickCount = 0; + #lastHour = -1; + #lastDay = -1; + + constructor(broadcast: ClockBroadcastFn, config?: Partial) { + this.#config = { ...DEFAULT_CONFIG, ...config }; + // Enforce minimum tick interval + if (this.#config.tickInterval < 10) this.#config.tickInterval = 10; + this.#broadcast = broadcast; + } + + /** Start the clock. Idempotent. */ + start(): void { + if (this.#timer || !this.#config.enabled) return; + + console.log( + `[SystemClock] Started (interval: ${this.#config.tickInterval}s, channels: ${this.#config.channels.join(", ")})`, + ); + + // Fire first tick immediately + this.#tick(); + + this.#timer = setInterval( + () => this.#tick(), + this.#config.tickInterval * 1000, + ); + } + + /** Stop the clock. */ + stop(): void { + if (this.#timer) { + clearInterval(this.#timer); + this.#timer = null; + console.log("[SystemClock] Stopped"); + } + } + + /** Check if the clock is running. */ + get running(): boolean { + return this.#timer !== null; + } + + /** Update config and restart if running. */ + updateConfig(config: Partial): void { + const wasRunning = this.running; + this.stop(); + this.#config = { ...this.#config, ...config }; + if (this.#config.tickInterval < 10) this.#config.tickInterval = 10; + if (wasRunning && this.#config.enabled) this.start(); + } + + #tick(): void { + this.#tickCount++; + const now = new Date(); + const payload = this.#makePayload(now); + + // Primary tick + if (this.#config.channels.includes("tick")) { + this.#broadcast("clock:tick", payload); + } + + // 5-minute tick + if ( + this.#config.channels.includes("minute:5") && + this.#tickCount > 1 && + now.getMinutes() % 5 === 0 && + now.getSeconds() < this.#config.tickInterval + ) { + this.#broadcast("clock:minute:5", payload); + } + + // Hourly tick + const currentHour = now.getUTCHours(); + if ( + this.#config.channels.includes("hourly") && + this.#lastHour !== -1 && + currentHour !== this.#lastHour + ) { + this.#broadcast("clock:hourly", payload); + } + this.#lastHour = currentHour; + + // Daily tick + const currentDay = now.getUTCDay(); + if ( + this.#config.channels.includes("daily") && + this.#lastDay !== -1 && + currentDay !== this.#lastDay + ) { + this.#broadcast("clock:daily", payload); + } + this.#lastDay = currentDay; + } + + #makePayload(now: Date): ClockPayload { + return { + timestamp: now.getTime(), + isoString: now.toISOString(), + hour: now.getUTCHours(), + minute: now.getUTCMinutes(), + second: now.getUTCSeconds(), + dayOfWeek: now.getUTCDay(), + tickCount: this.#tickCount, + }; + } +} diff --git a/server/index.ts b/server/index.ts index 8c92a7a..2e1e461 100644 --- a/server/index.ts +++ b/server/index.ts @@ -84,6 +84,8 @@ import { setNotionOAuthSyncServer } from "./oauth/notion"; import { setGoogleOAuthSyncServer } from "./oauth/google"; import { notificationRouter } from "./notification-routes"; import { registerUserConnection, unregisterUserConnection, notify } from "./notification-service"; +import { SystemClock } from "./clock-service"; +import type { ClockPayload } from "./clock-service"; // Register modules registerModule(canvasModule); @@ -2047,6 +2049,27 @@ function broadcastAutomergeSync(slug: string, excludePeerId?: string): void { } } +// ── System Clock ── + +/** Broadcast a clock event to all connected clients across all communities. */ +function broadcastClockEvent(channel: string, payload: ClockPayload): void { + const msg = JSON.stringify({ type: "clock", channel, payload }); + for (const [_slug, clients] of communityClients) { + for (const [_peerId, client] of clients) { + if (client.readyState === WebSocket.OPEN) { + try { + client.send(msg); + } catch { + // Ignore send errors on closing sockets + } + } + } + } +} + +const systemClock = new SystemClock(broadcastClockEvent); +systemClock.start(); + // ── Subdomain parsing (backward compat) ── const RESERVED_SUBDOMAINS = ["www", "rspace", "create", "new", "start", "auth"];