import { and, eq, isNull } from "drizzle-orm";
import type Redis from "ioredis";
import type { ServerEventEnvelope } from "@slackwsh/contracts";
import { schema } from "@slackwsh/data";
import { publishRoomEvent } from "./events-bus";

/** Every active workspace member — the fanout list for a workspace-wide
 * mutation (tasks, calendar events).
 *
 * Tasks and events are workspace-wide rather than channel-gated, and there is
 * no equivalent of `ch:{channelId}`'s "join on open" for a workspace room a
 * client can rely on always having joined (apps/gateway only joins
 * `ws:{workspaceId}` as a side effect of `channel:join`). So — mirroring
 * sendMessage's personal-fanout-to-peers pattern (§4.3) rather than `ws:`
 * room broadcast — these events go to every active member's own `u:{userId}`
 * room, which every connected socket joins unconditionally on connect. */
export async function activeWorkspaceMemberIds(tx: any, workspaceId: number): Promise<number[]> {
  const rows = await tx
    .select({ userId: schema.workspaceMembers.userId })
    .from(schema.workspaceMembers)
    .where(and(eq(schema.workspaceMembers.workspaceId, workspaceId), isNull(schema.workspaceMembers.deactivatedAt)));
  return rows.map((row: { userId: number }) => row.userId);
}

/** Publishes one envelope into every listed member's personal room.
 *
 * Concurrent, not sequential: publishRoomEvent never throws, and a workspace
 * of a few hundred members shouldn't cost a few hundred serialized Redis
 * round-trips on the mutation's response path. */
export async function fanoutToMembers(redis: Redis, memberIds: number[], event: ServerEventEnvelope): Promise<void> {
  await Promise.all(memberIds.map((memberId) => publishRoomEvent(redis, { room: `u:${memberId}`, event })));
}