task-stream/libs/api/tracker/infrastructure/src/lib/redis/redis-presence.store.ts

import { Inject, Injectable } from '@nestjs/common';
import { REDIS_CLIENT } from '@task-stream/api/shared/infrastructure';
import {
  PresenceEntry,
  PresenceStore,
} from '@task-stream/api/tracker/application';
import Redis from 'ioredis';

/**
 * One hash per ticket (field: user id, value: the entry). The hash expires
 * when nobody beats on it; stale fields are skipped and dropped on reading.
 */
@Injectable()
export class RedisPresenceStore extends PresenceStore {
  constructor(@Inject(REDIS_CLIENT) private readonly redis: Redis) {
    super();
  }

  private key(issueId: string): string {
    return `task-stream:presence:${issueId}`;
  }

  async touch(
    issueId: string,
    entry: PresenceEntry,
    ttlMs: number,
  ): Promise<void> {
    const key = this.key(issueId);
    await this.redis
      .multi()
      .hset(key, entry.user.id, JSON.stringify(entry))
      .pexpire(key, ttlMs)
      .exec();
  }

  async list(issueId: string, since: number): Promise<PresenceEntry[]> {
    const key = this.key(issueId);
    const all = await this.redis.hgetall(key);
    const fresh: PresenceEntry[] = [];
    const stale: string[] = [];
    for (const [userId, raw] of Object.entries(all)) {
      try {
        const entry = JSON.parse(raw) as PresenceEntry;
        if (Date.parse(entry.at) >= since) fresh.push(entry);
        else stale.push(userId);
      } catch {
        stale.push(userId);
      }
    }
    if (stale.length) await this.redis.hdel(key, ...stale);
    return fresh;
  }

  async leave(issueId: string, userId: string): Promise<void> {
    await this.redis.hdel(this.key(issueId), userId);
  }
}