@dudousxd/nestjs-agent-transport-redis

import { RedisTokenStreamSink } from '@dudousxd/nestjs-agent-transport-redis';
import { AgentModule, HeaderActorResolver } from '@dudousxd/nestjs-agent';

@Module({
  imports: [
    AgentModule.forRoot({
      model: myModelProvider,
      actorResolver: new HeaderActorResolver(),
      // myRedisClient: a RedisStreamClient (see below); options: { keyPrefix?: string; ttlSeconds?: number }
      sink: new RedisTokenStreamSink(myRedisClient), // shared across every replica
    }),
  ],
})
export class AppModule {}

interface RedisStreamClient {
  rpush(key: string, value: string): Promise<void>;
  lrange(key: string, start: number, stop: number): Promise<string[]>;
  set(key: string, value: string): Promise<void>;
  get(key: string): Promise<string | null>;
  publish(channel: string, message: string): Promise<void>;
  subscribe(channel: string, onMessage: (message: string) => void): Promise<() => Promise<void>>;
  del(...keys: string[]): Promise<void>;
  expire(key: string, seconds: number): Promise<void>;
}

// adapting ioredis: one command connection, one subscriber connection
import Redis from 'ioredis';
import type { RedisStreamClient } from '@dudousxd/nestjs-agent-transport-redis';

function ioredisStreamClient(redis: Redis, sub: Redis): RedisStreamClient {
  return {
    rpush: async (key, value) => { await redis.rpush(key, value); },
    lrange: (key, start, stop) => redis.lrange(key, start, stop),
    set: async (key, value) => { await redis.set(key, value); },
    get: (key) => redis.get(key),
    publish: async (channel, message) => { await redis.publish(channel, message); },
    subscribe: async (channel, onMessage) => {
      const listener = (ch: string, message: string) => { if (ch === channel) onMessage(message); };
      sub.on('message', listener);
      await sub.subscribe(channel);
      return async () => {
        sub.off('message', listener);
        await sub.unsubscribe(channel);
      };
    },
    del: async (...keys) => { await redis.del(...keys); },
    expire: async (key, seconds) => { await redis.expire(key, seconds); },
  };
}