Implement redis alarms using keyspace notifications

Author: zephraph

packages/drivers/redis/src/actor.ts

@@ -1,17 +1,45 @@
 import type { ActorDriver, KvKey, KvValue } from "actor-core/driver-helpers";
 import type Redis from "ioredis";
 import { KEYS } from "./keys";
-import { AnyActorInstance } from "actor-core/driver-helpers";
+import type { AnyActorInstance } from "actor-core/driver-helpers";
 
 export interface ActorDriverContext {
     redis: Redis;
 }
 
 export class RedisActorDriver implements ActorDriver {
     #redis: Redis;
+    #subscriptionRedis: Redis | null = null;
+    #alarmCallbacks = new Map<string, { actor: AnyActorInstance; timestamp: number }>();
 
     constructor(redis: Redis) {
         this.#redis = redis;
+
+        // Create a separate connection for subscriptions since a subscribed connection
+        // cannot be used for other commands
+        this.#subscriptionRedis = redis.duplicate();
+
+        // Subscribe to expired events
+        this.#subscriptionRedis.config('SET', 'notify-keyspace-events', 'Ex');
+        this.#subscriptionRedis.subscribe('__keyevent@0__:expired');
+
+        // Handle expired events
+        this.#subscriptionRedis.on('message', async (_channel, key) => {
+            // Extract actor ID from the key
+            const match = key.match(/^actor:(.+):alarm$/);
+            if (!match) return;
+
+            const actorId = match[1];
+            const callback = this.#alarmCallbacks.get(actorId);
+            if (callback) {
+                // Verify this is still the current alarm before triggering
+                const currentAlarm = await this.getAlarm(callback.actor);
+                if (currentAlarm === callback.timestamp) {
+                    await callback.actor.onAlarm();
+                    this.#alarmCallbacks.delete(actorId);
+                }
+            }
+        });
     }
 
     getContext(_actorId: string): ActorDriverContext {
@@ -52,8 +80,40 @@ export class RedisActorDriver implements ActorDriver {
         await this.#redis.del(key.map((k) => this.#serializeKey(actorId, k)));
     }
 
-    async setAlarm(_actor: AnyActorInstance, _timestamp: number): Promise<void> {
-        throw new Error("Alarms are not yet implemented for this driver.");
+    async setAlarm(actor: AnyActorInstance, timestamp: number): Promise<void> {
+        const key = KEYS.ACTOR.alarm(actor.id);
+
+        // Delete any existing alarm first
+        await this.deleteAlarm(actor);
+
+        const delay = timestamp - Date.now();
+        if (delay <= 0) {
+            // If timestamp is in the past, trigger immediately
+            await actor.onAlarm();
+            return;
+        }
+
+        // Store both the actor instance and timestamp for callback verification
+        this.#alarmCallbacks.set(actor.id, { actor, timestamp });
+
+        // Set the key with expiration
+        await this.#redis.set(key, timestamp.toString(), 'PX', delay);
+    }
+
+    async getAlarm(actor: AnyActorInstance): Promise<number | null> {
+        const key = KEYS.ACTOR.alarm(actor.id);
+
+        // Get the timestamp value
+        const value = await this.#redis.get(key);
+        if (!value) return null;
+
+        return Number.parseInt(value, 10);
+    }
+
+    async deleteAlarm(actor: AnyActorInstance): Promise<void> {
+        const key = KEYS.ACTOR.alarm(actor.id);
+        await this.#redis.del(key);
+        this.#alarmCallbacks.delete(actor.id);
     }
 
     #serializeKey(actorId: string, key: KvKey): string {

packages/drivers/redis/src/keys.ts

@@ -12,6 +12,8 @@ export const KEYS = {
 		tags: (actorId: string) => `actor:${actorId}:tags`,
 		// KEY
 		kv: (actorId: string, key: string) => `actor:${actorId}:kv:${key}`,
+		// KEY (sorted set) = timestamp
+		alarm: (actorId: string) => `actor:${actorId}:alarm`,
 	},
 };