import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from "@nestjs/common";
import { Inject } from "@nestjs/common";
import Redis from "ioredis";
import { and, eq } from "drizzle-orm";
import { schema, withTenant, withUser } from "@slackwsh/data";
import { asEntityId, type EntityIdInput, MESSAGING_EVENTS_CHANNEL, type RoomEvent } from "@slackwsh/messaging";
import { REDIS_PUBSUB } from "./redis-pubsub.provider";

const SOCKET_TTL_SECONDS = 60;
const SWEEP_INTERVAL_MS = 30_000;

export type PresenceStatus = "active" | "away" | "offline";

export interface PresenceRecord {
  status: PresenceStatus;
  lastSeen: string;
}

/**
 * Presence per ARCHITECTURE §4.4 — "the one piece Elixir gave for free."
 * Hand-built on Redis, deliberately best-effort and self-healing rather than
 * strongly consistent (never used for authorisation).
 *
 * Redis keys:
 *   presence:{workspaceId}     HASH   userId -> JSON {status, lastSeen}
 *   presence:sock:{socketId}   STRING userId, TTL 60s (liveness token)
 *   presence:user:{userId}     SET    socketIds currently held by this user
 *
 * Device count is derived via SCARD on `presence:user:{userId}` rather than
 * stored and incremented separately — self-correcting, can't drift out of
 * sync with the SET it's supposed to describe.
 *
 * Single-node caveat: the sweeper below runs as a plain `setInterval` in
 * this gateway process. That's correct for one node; a multi-node
 * deployment needs this to run as a single elected job (or in `apps/worker`)
 * so N gateway nodes don't all sweep the same keys redundantly — not built,
 * since this dev environment only ever runs one gateway node.
 */
@Injectable()
export class PresenceService implements OnModuleInit, OnModuleDestroy {
  private readonly logger = new Logger(PresenceService.name);
  private sweepTimer?: NodeJS.Timeout;

  constructor(@Inject(REDIS_PUBSUB) private readonly redis: Redis) {}

  onModuleInit() {
    this.sweepTimer = setInterval(() => this.sweep().catch((err) => this.logger.error(err)), SWEEP_INTERVAL_MS);
  }

  onModuleDestroy() {
    if (this.sweepTimer) clearInterval(this.sweepTimer);
  }

  /** Called on `channel:join` for a workspace not yet registered on this
   * socket. Returns true the first time this user comes online (deviceCount
   * 0 -> 1) — the caller broadcasts `presence:changed` only in that case, so
   * opening a second tab doesn't spam a redundant "still online" event. */
  async connect(workspaceId: EntityIdInput, userId: EntityIdInput, socketId: string): Promise<{ becameOnline: boolean }> {
    const wsId = asEntityId(workspaceId);
    const uId = asEntityId(userId);
    await this.redis.sadd(userSocketsKey(uId), socketId);
    await this.redis.set(socketLivenessKey(socketId), JSON.stringify({ userId: uId, active: true }), "EX", SOCKET_TTL_SECONDS);
    const deviceCount = await this.redis.scard(userSocketsKey(uId));

    await this.writeStatus(wsId, uId, await this.requestedStatus(wsId, uId, true));
    return { becameOnline: deviceCount === 1 };
  }

  async heartbeat(
    socketId: string,
    userId: EntityIdInput,
    workspaceIds: Iterable<EntityIdInput>,
    active: boolean,
  ): Promise<void> {
    const uId = asEntityId(userId);
    await this.redis.set(socketLivenessKey(socketId), JSON.stringify({ userId: uId, active }), "EX", SOCKET_TTL_SECONDS);
    const anyDeviceActive = await this.anyActiveSocket(uId);
    for (const workspaceId of workspaceIds) {
      const wsId = asEntityId(workspaceId);
      await this.writeStatus(wsId, uId, await this.requestedStatus(wsId, uId, anyDeviceActive));
    }
  }

  /** Only publish `offline` when deviceCount reaches zero — closing one tab
   * must not mark a user offline while they're still active elsewhere. This
   * is the specific defect §4.4 exists to prevent. */
  async disconnect(workspaceId: EntityIdInput, userId: EntityIdInput, socketId: string): Promise<{ becameOffline: boolean }> {
    const wsId = asEntityId(workspaceId);
    const uId = asEntityId(userId);
    await this.redis.srem(userSocketsKey(uId), socketId);
    await this.redis.del(socketLivenessKey(socketId));
    const deviceCount = await this.redis.scard(userSocketsKey(uId));

    if (deviceCount === 0) {
      await this.writeStatus(wsId, uId, "offline");
      return { becameOffline: true };
    }
    await this.writeStatus(wsId, uId, await this.requestedStatus(wsId, uId, await this.anyActiveSocket(uId)));
    return { becameOffline: false };
  }

  async getWorkspacePresence(workspaceId: EntityIdInput): Promise<Record<string, PresenceRecord>> {
    const raw = await this.redis.hgetall(presenceHashKey(asEntityId(workspaceId)));
    const result: Record<string, PresenceRecord> = {};
    for (const [userId, json] of Object.entries(raw)) {
      try {
        result[userId] = JSON.parse(json);
      } catch {
        // malformed entry — skip rather than fail the whole read
      }
    }
    return result;
  }

  private async writeStatus(workspaceId: number, userId: number, status: PresenceStatus): Promise<void> {
    const record: PresenceRecord = { status, lastSeen: new Date().toISOString() };
    const previousRaw = await this.redis.hget(presenceHashKey(workspaceId), String(userId));
    let previousStatus: PresenceStatus | null = null;
    try {
      previousStatus = previousRaw ? (JSON.parse(previousRaw) as PresenceRecord).status : null;
    } catch {
      previousStatus = null;
    }
    await this.redis.hset(presenceHashKey(workspaceId), String(userId), JSON.stringify(record));

    if (previousStatus === status) return;

    const event: RoomEvent = {
      room: `ws:${workspaceId}`,
      event: { type: "presence:changed", payload: { workspaceId, userId, status } },
    };
    await this.redis.publish(MESSAGING_EVENTS_CHANNEL, JSON.stringify(event)).catch((err) => this.logger.warn(String(err)));
  }

  private async requestedStatus(workspaceId: number, userId: number, active: boolean): Promise<PresenceStatus> {
    const mode = await withTenant({ workspaceId, userId }, async (tx) => {
      const [member] = await tx
        .select({ availabilityMode: schema.workspaceMembers.availabilityMode })
        .from(schema.workspaceMembers)
        .where(and(eq(schema.workspaceMembers.workspaceId, workspaceId), eq(schema.workspaceMembers.userId, userId)))
        .limit(1);
      return member?.availabilityMode;
    }).catch(() => "auto");
    return mode === "away" ? "away" : active ? "active" : "away";
  }

  /** Reconciles `presence:user:{userId}` sets against surviving TTL tokens —
   * covers a gateway process dying ungracefully (no disconnect event ever
   * fires) so I9's "shown offline within 60s" holds even then. Iterates via
   * SCAN rather than KEYS to avoid blocking Redis on a large keyspace. */
  private async sweep(): Promise<void> {
    let cursor = "0";
    do {
      const [next, keys] = await this.redis.scan(cursor, "MATCH", "presence:user:*", "COUNT", 100);
      cursor = next;
      for (const key of keys) {
        const rawUserId = key.slice("presence:user:".length);
        let userId: number;
        try {
          userId = asEntityId(rawUserId);
        } catch {
          continue;
        }
        const socketIds = await this.redis.smembers(key);
        if (socketIds.length === 0) continue;

        const alive = await Promise.all(socketIds.map((id) => this.redis.exists(socketLivenessKey(id))));
        const stale = socketIds.filter((_, i) => !alive[i]);
        if (stale.length === 0) continue;

        await this.redis.srem(key, ...stale);
        const remaining = await this.redis.scard(key);
        if (remaining === 0) {
          // No live sockets survive for this user anywhere — mark them
          // offline in every workspace they belong to. Covers a gateway
          // process dying ungracefully (no disconnect event ever fires),
          // which is the specific case I9's "within 60s" bound exists for.
          const memberships = await withUser(userId, (tx) =>
            tx
              .select({ workspaceId: schema.workspaceMembers.workspaceId })
              .from(schema.workspaceMembers)
              .where(eq(schema.workspaceMembers.userId, userId)),
          );
          for (const { workspaceId } of memberships) {
            await this.writeStatus(workspaceId, userId, "offline");
          }
        } else {
          const anyDeviceActive = await this.anyActiveSocket(userId);
          const memberships = await withUser(userId, (tx) =>
            tx.select({ workspaceId: schema.workspaceMembers.workspaceId }).from(schema.workspaceMembers).where(eq(schema.workspaceMembers.userId, userId)),
          );
          for (const { workspaceId } of memberships) {
            await this.writeStatus(workspaceId, userId, await this.requestedStatus(workspaceId, userId, anyDeviceActive));
          }
        }
      }
    } while (cursor !== "0");
  }

  private async anyActiveSocket(userId: number): Promise<boolean> {
    const socketIds = await this.redis.smembers(userSocketsKey(userId));
    if (socketIds.length === 0) return false;
    const records = await Promise.all(socketIds.map((socketId) => this.redis.get(socketLivenessKey(socketId))));
    return records.some((raw) => {
      if (!raw) return false;
      try {
        const parsed = JSON.parse(raw) as { active?: boolean };
        return parsed.active !== false;
      } catch {
        // Legacy liveness tokens held only the user id and represented an active socket.
        return true;
      }
    });
  }
}

function presenceHashKey(workspaceId: number): string {
  return `presence:${workspaceId}`;
}

function socketLivenessKey(socketId: string): string {
  return `presence:sock:${socketId}`;
}

function userSocketsKey(userId: number): string {
  return `presence:user:${userId}`;
}