import { ForbiddenException, BadRequestException, NotFoundException } from "@nestjs/common";
import { and, desc, eq, inArray, isNull, sql } from "drizzle-orm";
import type Redis from "ioredis";
import type { Channel } from "@slackwsh/contracts";
import { schema, withTenant } from "@slackwsh/data";
import { can } from "@slackwsh/core";
import { publishRoomEvent } from "./events-bus";

export type EntityIdInput = number | string;

export function asEntityId(value: EntityIdInput): number {
  const n = typeof value === "number" ? value : Number(value);
  if (!Number.isInteger(n) || n <= 0) throw new BadRequestException("invalid id");
  return n;
}

/**
 * Every member-scoped action fetches the actor's own workspace membership
 * row first — never trust a client-supplied role. Shared with
 * apps/api/src/workspaces/workspaces.service.ts's pattern.
 */
export async function requireWorkspaceMembership(tx: any, workspaceId: EntityIdInput, userId: EntityIdInput) {
  const wsId = asEntityId(workspaceId);
  const uId = asEntityId(userId);
  const [row] = await tx
    .select()
    .from(schema.workspaceMembers)
    .where(and(eq(schema.workspaceMembers.workspaceId, wsId), eq(schema.workspaceMembers.userId, uId)))
    .limit(1);
  if (!row || row.deactivatedAt) throw new ForbiddenException("not an active member of this workspace");
  return row;
}

export async function requireChannelMembership(tx: any, channelId: EntityIdInput, userId: EntityIdInput) {
  const chId = asEntityId(channelId);
  const uId = asEntityId(userId);
  const [row] = await tx
    .select()
    .from(schema.channelMembers)
    .where(and(eq(schema.channelMembers.channelId, chId), eq(schema.channelMembers.userId, uId)))
    .limit(1);
  if (!row) throw new ForbiddenException("not a member of this channel");
  return row;
}

export async function createChannel(
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  input: { name: string; type: "public" | "private"; topic?: string },
) {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    const membership = await requireWorkspaceMembership(tx, wsId, userId);
    if (!can({ userId, role: membership.role }, "channel:create")) {
      throw new ForbiddenException("insufficient role to create channels");
    }
    const normalizedName = input.name.trim();
    const [duplicate] = await tx
      .select({ id: schema.channels.id })
      .from(schema.channels)
      .where(and(eq(schema.channels.workspaceId, wsId), sql`lower(${schema.channels.name}) = lower(${normalizedName})`))
      .limit(1);
    if (duplicate) throw new BadRequestException("a channel with that name already exists");

    const [channel] = await tx
      .insert(schema.channels)
      .values({
        workspaceId: wsId,
        type: input.type,
        name: normalizedName,
        topic: input.topic ?? null,
        createdBy: userId,
        memberCount: 1,
      })
      .returning();
    if (!channel) throw new Error("channel insert returned no row");

    await tx.insert(schema.channelSeq).values({ channelId: channel.id, lastSeq: 0 });
    await tx.insert(schema.channelMembers).values({ channelId: channel.id, userId, role: "owner" });

    return channel;
  });
}

export async function listMyChannels(workspaceId: EntityIdInput, actorUserId: EntityIdInput) {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    await requireWorkspaceMembership(tx, wsId, userId);
    const rows = await tx
      .select({
        channel: schema.channels,
        member: schema.channelMembers,
        // I5 / ADR-014: thread replies share the channel seq space but do not
        // count toward channel unread unless also sent to the channel.
        unreadCount: sql<number>`(
          select count(*)::integer
          from messages unread_messages
          where unread_messages.channel_id = ${schema.channels.id}
            and unread_messages.seq > ${schema.channelMembers.lastReadSeq}
            and (unread_messages.parent_id is null or unread_messages.is_broadcast)
            and unread_messages.deleted_at is null
            and unread_messages.author_id <> ${userId}
        )`,
      })
      .from(schema.channelMembers)
      .innerJoin(schema.channels, eq(schema.channels.id, schema.channelMembers.channelId))
      .where(and(eq(schema.channels.workspaceId, wsId), eq(schema.channelMembers.userId, userId)));

    // Attach DM peers so the client can label 1:1 / group DMs with people
    // names instead of falling back to a blank "#channel" title.
    const enriched = [];
    for (const row of rows) {
      if (row.channel.type !== "dm" && row.channel.type !== "group_dm") {
        enriched.push({
          channel: row.channel,
          member: { ...row.member, unreadCount: Number(row.unreadCount) },
          dmPeer: null as null | { id: number; name: string; email: string; avatarUrl: string | null },
          dmPeers: [] as Array<{ id: number; name: string; email: string; avatarUrl: string | null }>,
        });
        continue;
      }
      const peers = await tx
        .select({
          id: schema.users.id,
          name: schema.users.name,
          email: schema.users.email,
          avatarUrl: schema.users.avatarUrl,
        })
        .from(schema.channelMembers)
        .innerJoin(schema.users, eq(schema.users.id, schema.channelMembers.userId))
        .where(and(eq(schema.channelMembers.channelId, row.channel.id)));
      const others = peers.filter((p) => p.id !== userId);
      enriched.push({
        channel: row.channel,
        member: { ...row.member, unreadCount: Number(row.unreadCount) },
        dmPeer: others[0] ?? null,
        dmPeers: others,
      });
    }
    return enriched;
  });
}

export async function joinChannel(workspaceId: EntityIdInput, actorUserId: EntityIdInput, channelId: EntityIdInput) {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  const chId = asEntityId(channelId);
  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    await requireWorkspaceMembership(tx, wsId, userId);
    const [channel] = await tx.select().from(schema.channels).where(eq(schema.channels.id, chId)).limit(1);
    if (!channel || channel.workspaceId !== wsId) throw new NotFoundException("channel not found");
    if (channel.type !== "public") throw new ForbiddenException("only public channels can be self-joined");

    const existing = await tx
      .select()
      .from(schema.channelMembers)
      .where(and(eq(schema.channelMembers.channelId, chId), eq(schema.channelMembers.userId, userId)))
      .limit(1);
    if (existing.length === 0) {
      await tx.insert(schema.channelMembers).values({ channelId: chId, userId, role: "member" });
      await tx
        .update(schema.channels)
        .set({ memberCount: sql`${schema.channels.memberCount} + 1` })
        .where(eq(schema.channels.id, chId));
    }
    return channel;
  });
}

export async function leaveChannel(workspaceId: EntityIdInput, actorUserId: EntityIdInput, channelId: EntityIdInput) {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  const chId = asEntityId(channelId);
  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    await requireChannelMembership(tx, chId, userId);
    const [channel] = await tx.select().from(schema.channels).where(eq(schema.channels.id, chId)).limit(1);
    if (!channel || channel.workspaceId !== wsId) throw new NotFoundException("channel not found");
    // A DM has no "membership" to give up — leaving one would leave the other
    // participant talking to a room nobody can rejoin (there is no join path
    // back into a dm/group_dm). Closing the conversation is a client concern.
    if (channel.type === "dm" || channel.type === "group_dm") {
      throw new BadRequestException("cannot leave a direct message");
    }
    if (channel.name?.toLowerCase() === "general") {
      throw new BadRequestException("members cannot leave #general");
    }
    await tx
      .delete(schema.channelMembers)
      .where(and(eq(schema.channelMembers.channelId, chId), eq(schema.channelMembers.userId, userId)));
    await tx
      .update(schema.channels)
      .set({ memberCount: sql`greatest(${schema.channels.memberCount} - 1, 0)` })
      .where(eq(schema.channels.id, chId));
  });
}

/** Add existing workspace members into a text channel. Actor must already be
 * in the channel; targets must be active workspace members (not guests from
 * outside the workspace — those join via workspace invite first). */
export async function addChannelMembers(
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  channelId: EntityIdInput,
  userIds: EntityIdInput[],
) {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  const chId = asEntityId(channelId);
  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    await requireWorkspaceMembership(tx, wsId, userId);
    await requireChannelMembership(tx, chId, userId);

    const [channel] = await tx.select().from(schema.channels).where(eq(schema.channels.id, chId)).limit(1);
    if (!channel || channel.workspaceId !== wsId) throw new NotFoundException("channel not found");
    if (channel.type === "dm" || channel.type === "group_dm") {
      throw new BadRequestException("cannot add members to a direct message this way");
    }

    const unique = [...new Set(userIds.map(asEntityId))].filter((id) => id !== userId);
    if (unique.length === 0) return { added: [] as number[] };

    const eligible = await tx
      .select({ userId: schema.workspaceMembers.userId })
      .from(schema.workspaceMembers)
      .where(
        and(
          eq(schema.workspaceMembers.workspaceId, wsId),
          inArray(schema.workspaceMembers.userId, unique),
          isNull(schema.workspaceMembers.deactivatedAt),
        ),
      );
    const eligibleIds = eligible.map((row) => row.userId);
    if (eligibleIds.length === 0) return { added: [] as number[] };

    const already = await tx
      .select({ userId: schema.channelMembers.userId })
      .from(schema.channelMembers)
      .where(and(eq(schema.channelMembers.channelId, chId), inArray(schema.channelMembers.userId, eligibleIds)));
    const alreadySet = new Set(already.map((row) => row.userId));
    const toAdd = eligibleIds.filter((id) => !alreadySet.has(id));
    if (toAdd.length === 0) return { added: [] as number[] };

    await tx.insert(schema.channelMembers).values(toAdd.map((uid) => ({ channelId: chId, userId: uid, role: "member" })));
    await tx
      .update(schema.channels)
      .set({ memberCount: sql`${schema.channels.memberCount} + ${toAdd.length}` })
      .where(eq(schema.channels.id, chId));

    return { added: toAdd };
  });
}

export async function listChannelMembers(workspaceId: EntityIdInput, actorUserId: EntityIdInput, channelId: EntityIdInput) {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  const chId = asEntityId(channelId);
  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    await requireWorkspaceMembership(tx, wsId, userId);
    await requireChannelMembership(tx, chId, userId);

    const [channel] = await tx.select().from(schema.channels).where(eq(schema.channels.id, chId)).limit(1);
    if (!channel || channel.workspaceId !== wsId) throw new NotFoundException("channel not found");

    const rows = await tx
      .select({
        member: schema.channelMembers,
        user: {
          id: schema.users.id,
          name: schema.users.name,
          email: schema.users.email,
          avatarUrl: schema.users.avatarUrl,
        },
      })
      .from(schema.channelMembers)
      .innerJoin(schema.users, eq(schema.users.id, schema.channelMembers.userId))
      .where(eq(schema.channelMembers.channelId, chId));

    return rows;
  });
}

/** Removes one person from a text channel without removing them from the
 * workspace. Workspace admins or the channel owner may do this. */
export async function removeChannelMember(
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  channelId: EntityIdInput,
  memberUserId: EntityIdInput,
) {
  const wsId = asEntityId(workspaceId);
  const actorId = asEntityId(actorUserId);
  const chId = asEntityId(channelId);
  const targetId = asEntityId(memberUserId);
  if (actorId === targetId) throw new BadRequestException("use leave channel to remove yourself");

  return withTenant({ workspaceId: wsId, userId: actorId }, async (tx) => {
    const workspaceMembership = await requireWorkspaceMembership(tx, wsId, actorId);
    const actorMembership = await requireChannelMembership(tx, chId, actorId);
    const [channel] = await tx.select().from(schema.channels).where(eq(schema.channels.id, chId)).limit(1);
    if (!channel || channel.workspaceId !== wsId) throw new NotFoundException("channel not found");
    if (channel.type !== "public" && channel.type !== "private") {
      throw new BadRequestException("members cannot be removed from direct messages");
    }
    if (channel.name?.toLowerCase() === "general") {
      throw new BadRequestException("members cannot be removed from #general");
    }
    if (!can({ userId: actorId, role: workspaceMembership.role }, "member:remove") && actorMembership.role !== "owner") {
      throw new ForbiddenException("only a channel owner or workspace admin can remove members");
    }

    const [target] = await tx
      .select()
      .from(schema.channelMembers)
      .where(and(eq(schema.channelMembers.channelId, chId), eq(schema.channelMembers.userId, targetId)))
      .limit(1);
    if (!target) throw new NotFoundException("channel member not found");
    if (target.role === "owner" && workspaceMembership.role !== "owner") {
      throw new ForbiddenException("only the workspace owner can remove a channel owner");
    }

    await tx
      .delete(schema.channelMembers)
      .where(and(eq(schema.channelMembers.channelId, chId), eq(schema.channelMembers.userId, targetId)));
    await tx
      .update(schema.channels)
      .set({ memberCount: sql`greatest(${schema.channels.memberCount} - 1, 0)` })
      .where(eq(schema.channels.id, chId));
    return { removed: targetId };
  });
}

/** Irreversibly removes a text channel and its conversation data. Workspace-
 * wide tasks/events/calls are preserved but detached from the deleted room. */
export async function permanentlyDeleteChannel(
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  channelId: EntityIdInput,
) {
  const wsId = asEntityId(workspaceId);
  const actorId = asEntityId(actorUserId);
  const chId = asEntityId(channelId);
  return withTenant({ workspaceId: wsId, userId: actorId }, async (tx) => {
    const membership = await requireWorkspaceMembership(tx, wsId, actorId);
    if (membership.role !== "owner") throw new ForbiddenException("only the workspace owner can permanently delete channels");
    const [channel] = await tx.select().from(schema.channels).where(eq(schema.channels.id, chId)).limit(1);
    if (!channel || channel.workspaceId !== wsId) throw new NotFoundException("channel not found");
    if (channel.type !== "public" && channel.type !== "private") {
      throw new BadRequestException("direct messages cannot be permanently deleted");
    }
    if (["general", "random"].includes(channel.name?.toLowerCase() ?? "")) {
      throw new BadRequestException("default channels cannot be permanently deleted");
    }

    const attachments = await tx
      .select({ objectKey: schema.attachments.objectKey })
      .from(schema.attachments)
      .where(eq(schema.attachments.channelId, chId));
    const messageRows = await tx
      .select({ id: schema.messages.id })
      .from(schema.messages)
      .where(eq(schema.messages.channelId, chId));
    const messageIds = messageRows.map((row) => row.id);

    await tx.update(schema.tasks).set({ channelId: null, sourceMessageId: null }).where(eq(schema.tasks.channelId, chId));
    await tx.update(schema.calls).set({ channelId: null }).where(eq(schema.calls.channelId, chId));
    await tx.update(schema.events).set({ channelId: null }).where(eq(schema.events.channelId, chId));
    await tx.delete(schema.threadSubscriptions).where(eq(schema.threadSubscriptions.channelId, chId));
    await tx.delete(schema.pins).where(eq(schema.pins.channelId, chId));
    if (messageIds.length > 0) {
      await tx.delete(schema.reactions).where(inArray(schema.reactions.messageId, messageIds));
      await tx.delete(schema.messageMentions).where(inArray(schema.messageMentions.messageId, messageIds));
      await tx
        .update(schema.tasks)
        .set({ sourceMessageId: null })
        .where(inArray(schema.tasks.sourceMessageId, messageIds));
    }
    await tx.delete(schema.attachments).where(eq(schema.attachments.channelId, chId));
    await tx.delete(schema.messages).where(eq(schema.messages.channelId, chId));
    await tx.delete(schema.channelIntegrations).where(eq(schema.channelIntegrations.channelId, chId));
    await tx.delete(schema.channelMembers).where(eq(schema.channelMembers.channelId, chId));
    await tx.delete(schema.channelSeq).where(eq(schema.channelSeq.channelId, chId));
    await tx.delete(schema.channels).where(eq(schema.channels.id, chId));
    return { deleted: true, objectKeys: attachments.map((row) => row.objectKey) };
  });
}

export async function archiveChannel(
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  channelId: EntityIdInput,
  archived: boolean,
) {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  const chId = asEntityId(channelId);
  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    const membership = await requireWorkspaceMembership(tx, wsId, userId);
    if (!can({ userId, role: membership.role }, "channel:archive")) {
      throw new ForbiddenException("insufficient role to archive channels");
    }
    // The channel itself is the more-specific resource being mutated — fetch
    // and check it belongs to this workspace before writing, rather than
    // trusting the two ids (workspaceId, channelId) are consistent together.
    const [channel] = await tx.select().from(schema.channels).where(eq(schema.channels.id, chId)).limit(1);
    if (!channel || channel.workspaceId !== wsId) throw new NotFoundException("channel not found");
    if (["general", "random"].includes(channel.name?.toLowerCase() ?? "")) {
      throw new BadRequestException("default channels cannot be archived");
    }

    await tx.update(schema.channels).set({ isArchived: archived }).where(eq(schema.channels.id, chId));
  });
}

/** The wire shape of a channel (§ entities.Channel) — dates as ISO strings. */
function toWireChannel(row: typeof schema.channels.$inferSelect): Channel {
  return {
    id: row.id,
    workspaceId: row.workspaceId,
    type: row.type,
    name: row.name,
    topic: row.topic,
    purpose: row.purpose,
    createdBy: row.createdBy,
    isArchived: row.isArchived,
    lastMessageAt: row.lastMessageAt?.toISOString() ?? null,
    memberCount: row.memberCount,
  } as Channel;
}

/**
 * Renames/updates topic/purpose — Feature 3.5. `purpose` is what the About
 * panel shows as the room description; `topic` stays the short header line.
 *
 * Publishes `channel:updated` so a description edited in one client shows up
 * in everyone else's About panel without a reload — the same full-row event
 * shape archiveChannel's consumers already expect.
 */
export async function updateChannel(
  redis: Redis | null,
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  channelId: EntityIdInput,
  input: { name?: string; topic?: string | null; purpose?: string | null },
) {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  const chId = asEntityId(channelId);
  const updated = await withTenant({ workspaceId: wsId, userId }, async (tx) => {
    const membership = await requireWorkspaceMembership(tx, wsId, userId);
    await requireChannelMembership(tx, chId, userId);
    if (!can({ userId, role: membership.role }, "channel:update")) {
      throw new ForbiddenException("insufficient role to update this room");
    }
    const [channel] = await tx.select().from(schema.channels).where(eq(schema.channels.id, chId)).limit(1);
    if (!channel || channel.workspaceId !== wsId) throw new NotFoundException("channel not found");
    if (input.name !== undefined && ["general", "random"].includes(channel.name?.toLowerCase() ?? "")) {
      throw new BadRequestException("default channels cannot be renamed");
    }
    if (input.name !== undefined && channel.type === "dm") {
      throw new BadRequestException("a one-to-one direct message cannot be renamed");
    }
    if (input.name !== undefined && channel.type !== "group_dm") {
      const [duplicate] = await tx
        .select({ id: schema.channels.id })
        .from(schema.channels)
        .where(
          and(
            eq(schema.channels.workspaceId, wsId),
            sql`${schema.channels.id} <> ${chId}`,
            sql`lower(${schema.channels.name}) = lower(${input.name.trim()})`,
          ),
        )
        .limit(1);
      if (duplicate) throw new BadRequestException("a channel with that name already exists");
    }

    const patch: Record<string, unknown> = {};
    if (input.name !== undefined) patch.name = input.name;
    // Empty string means "clear it" — stored as NULL so the panel falls back
    // to its placeholder rather than rendering a blank description.
    if (input.topic !== undefined) patch.topic = input.topic?.trim() ? input.topic.trim() : null;
    if (input.purpose !== undefined) patch.purpose = input.purpose?.trim() ? input.purpose.trim() : null;
    if (Object.keys(patch).length === 0) return channel;

    const [next] = await tx.update(schema.channels).set(patch).where(eq(schema.channels.id, chId)).returning();
    return next ?? channel;
  });

  if (redis) {
    await publishRoomEvent(redis, {
      room: `ch:${chId}`,
      event: { type: "channel:updated", payload: { channel: toWireChannel(updated) } },
    });
  }
  return toWireChannel(updated);
}

/**
 * Per-user notification settings for one room. Deliberately scoped to the
 * caller's own channel_members row — there is no "set someone else's
 * preference" path, so no role check is needed beyond being in the room.
 */
export async function updateChannelPrefs(
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  channelId: EntityIdInput,
  input: { notifPref?: "all" | "mentions" | "none"; isMuted?: boolean; isStarred?: boolean },
) {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  const chId = asEntityId(channelId);
  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    await requireChannelMembership(tx, chId, userId);

    const patch: Record<string, unknown> = {};
    if (input.notifPref !== undefined) patch.notifPref = input.notifPref;
    if (input.isMuted !== undefined) patch.isMuted = input.isMuted;
    if (input.isStarred !== undefined) patch.isStarred = input.isStarred;
    if (Object.keys(patch).length === 0) throw new BadRequestException("no preference fields supplied");

    const [updated] = await tx
      .update(schema.channelMembers)
      .set(patch)
      .where(and(eq(schema.channelMembers.channelId, chId), eq(schema.channelMembers.userId, userId)))
      .returning();
    return updated!;
  });
}

/** Links an app into a room. Re-linking the same provider updates it in
 * place — the unique (channel_id, provider) index is what makes that an
 * upsert rather than a duplicate row. */
export async function addChannelIntegration(
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  channelId: EntityIdInput,
  input: { provider: string; label?: string; externalUrl?: string },
) {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  const chId = asEntityId(channelId);
  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    const membership = await requireWorkspaceMembership(tx, wsId, userId);
    await requireChannelMembership(tx, chId, userId);
    if (!can({ userId, role: membership.role }, "channel:update")) {
      throw new ForbiddenException("insufficient role to change this room's apps");
    }
    const [channel] = await tx.select().from(schema.channels).where(eq(schema.channels.id, chId)).limit(1);
    if (!channel || channel.workspaceId !== wsId) throw new NotFoundException("channel not found");

    const [row] = await tx
      .insert(schema.channelIntegrations)
      .values({
        channelId: chId,
        provider: input.provider,
        label: input.label ?? null,
        externalUrl: input.externalUrl ?? null,
        addedBy: userId,
      })
      .onConflictDoUpdate({
        target: [schema.channelIntegrations.channelId, schema.channelIntegrations.provider],
        set: { label: input.label ?? null, externalUrl: input.externalUrl ?? null, addedBy: userId },
      })
      .returning();
    return row!;
  });
}

export async function removeChannelIntegration(
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  channelId: EntityIdInput,
  integrationId: EntityIdInput,
) {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  const chId = asEntityId(channelId);
  const intId = asEntityId(integrationId);
  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    const membership = await requireWorkspaceMembership(tx, wsId, userId);
    await requireChannelMembership(tx, chId, userId);
    if (!can({ userId, role: membership.role }, "channel:update")) {
      throw new ForbiddenException("insufficient role to change this room's apps");
    }
    // Scope the delete by channel as well as id: the id alone is enough for
    // RLS (same workspace) but not enough to stop one room's member removing
    // another room's integration.
    await tx
      .delete(schema.channelIntegrations)
      .where(and(eq(schema.channelIntegrations.id, intId), eq(schema.channelIntegrations.channelId, chId)));
  });
}

export interface ChannelAboutPin {
  messageId: number;
  text: string;
  authorId: number;
  authorName: string;
  pinnedBy: number;
  pinnedByName: string;
  pinnedAt: string;
  createdAt: string;
}

/**
 * Everything the About-this-room panel renders, in one round trip: the room
 * row itself, the caller's own membership (role + notification prefs), the
 * pinned messages, and the linked apps. One endpoint rather than four because
 * the panel is all-or-nothing — it opens as a unit.
 */
export async function getChannelAbout(workspaceId: EntityIdInput, actorUserId: EntityIdInput, channelId: EntityIdInput) {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  const chId = asEntityId(channelId);
  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    const workspaceMembership = await requireWorkspaceMembership(tx, wsId, userId);
    const membership = await requireChannelMembership(tx, chId, userId);
    const [channel] = await tx.select().from(schema.channels).where(eq(schema.channels.id, chId)).limit(1);
    if (!channel || channel.workspaceId !== wsId) throw new NotFoundException("channel not found");

    const pinRows = await tx
      .select({
        messageId: schema.pins.messageId,
        pinnedBy: schema.pins.pinnedBy,
        pinnedAt: schema.pins.pinnedAt,
        text: schema.messages.text,
        authorId: schema.messages.authorId,
        deletedAt: schema.messages.deletedAt,
        createdAt: schema.messages.createdAt,
      })
      .from(schema.pins)
      .innerJoin(schema.messages, eq(schema.messages.id, schema.pins.messageId))
      .where(eq(schema.pins.channelId, chId))
      .orderBy(desc(schema.pins.pinnedAt));

    const live = pinRows.filter((row) => row.deletedAt === null);
    const nameIds = [...new Set(live.flatMap((row) => [row.authorId, row.pinnedBy]))];
    const nameRows =
      nameIds.length > 0
        ? await tx
            .select({ id: schema.users.id, name: schema.users.name })
            .from(schema.users)
            .where(inArray(schema.users.id, nameIds))
        : [];
    const nameById = new Map(nameRows.map((row) => [row.id, row.name]));

    const pins: ChannelAboutPin[] = live.map((row) => ({
      messageId: row.messageId,
      text: row.text,
      authorId: row.authorId,
      authorName: nameById.get(row.authorId) ?? "Unknown",
      pinnedBy: row.pinnedBy,
      pinnedByName: nameById.get(row.pinnedBy) ?? "Unknown",
      pinnedAt: row.pinnedAt.toISOString(),
      createdAt: row.createdAt.toISOString(),
    }));

    const integrations = await tx
      .select()
      .from(schema.channelIntegrations)
      .where(eq(schema.channelIntegrations.channelId, chId))
      .orderBy(desc(schema.channelIntegrations.addedAt));

    return {
      channel: toWireChannel(channel),
      membership: {
        role: membership.role,
        notifPref: membership.notifPref,
        isMuted: membership.isMuted,
        isStarred: membership.isStarred,
        joinedAt: membership.joinedAt.toISOString(),
      },
      // What *this* caller may do, resolved server-side so the panel doesn't
      // re-implement the policy table and then disagree with it.
      capabilities: {
        canEditRoom: can({ userId, role: workspaceMembership.role }, "channel:update"),
        canPin: can({ userId, role: workspaceMembership.role }, "message:pin"),
        canArchive: can({ userId, role: workspaceMembership.role }, "channel:archive"),
        canLeave: channel.type === "public" || channel.type === "private",
        canRemoveMembers:
          can({ userId, role: workspaceMembership.role }, "member:remove") || membership.role === "owner",
        canDeletePermanently: workspaceMembership.role === "owner",
      },
      pins,
      integrations: integrations.map((row) => ({
        id: row.id,
        provider: row.provider,
        label: row.label,
        externalUrl: row.externalUrl,
        addedBy: row.addedBy,
        addedAt: row.addedAt.toISOString(),
      })),
    };
  });
}

/** DMs are channels of type `dm` with exactly the two participants (§4.7). */
export async function getOrCreateDirectMessage(
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  otherUserId: EntityIdInput,
) {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  const otherId = asEntityId(otherUserId);
  if (userId === otherId) throw new BadRequestException("cannot open a DM with yourself");

  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    await requireWorkspaceMembership(tx, wsId, userId);
    await requireWorkspaceMembership(tx, wsId, otherId);

    const mine = await tx
      .select({ channelId: schema.channelMembers.channelId })
      .from(schema.channelMembers)
      .innerJoin(schema.channels, eq(schema.channels.id, schema.channelMembers.channelId))
      .where(and(eq(schema.channelMembers.userId, userId), eq(schema.channels.type, "dm"), eq(schema.channels.workspaceId, wsId)));

    if (mine.length > 0) {
      const theirs = await tx
        .select({ channelId: schema.channelMembers.channelId })
        .from(schema.channelMembers)
        .where(
          and(
            eq(schema.channelMembers.userId, otherId),
            inArray(
              schema.channelMembers.channelId,
              mine.map((m) => m.channelId),
            ),
          ),
        );
      if (theirs.length > 0) {
        const [existing] = await tx.select().from(schema.channels).where(eq(schema.channels.id, theirs[0]!.channelId)).limit(1);
        if (existing) {
          await tx
            .update(schema.channelMembers)
            .set({ isClosed: false })
            .where(and(eq(schema.channelMembers.channelId, existing.id), eq(schema.channelMembers.userId, userId)));
          return existing;
        }
      }
    }

    const [channel] = await tx
      .insert(schema.channels)
      .values({ workspaceId: wsId, type: "dm", createdBy: userId, memberCount: 2 })
      .returning();
    if (!channel) throw new Error("channel insert returned no row");

    await tx.insert(schema.channelSeq).values({ channelId: channel.id, lastSeq: 0 });
    await tx.insert(schema.channelMembers).values([
      { channelId: channel.id, userId, role: "member" },
      { channelId: channel.id, userId: otherId, role: "member" },
    ]);

    return channel;
  });
}

/**
 * Group DMs (Feature 3.4, up to 9 participants). Unlike 1:1 DMs, this does
 * not dedupe against an existing channel with the same member set — Slack
 * itself allows creating a fresh group DM for the same set, and computing
 * "the" canonical channel for an arbitrary member set adds real complexity
 * for a Phase 2 MVP. Documented simplification, not an oversight.
 */
export async function createGroupDirectMessage(
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  memberUserIds: EntityIdInput[],
) {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  const uniqueMembers = Array.from(new Set([userId, ...memberUserIds.map(asEntityId)]));
  if (uniqueMembers.length < 3) throw new BadRequestException("group DMs need at least 3 participants");
  if (uniqueMembers.length > 9) throw new BadRequestException("group DMs support at most 9 participants");

  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    for (const memberId of uniqueMembers) {
      await requireWorkspaceMembership(tx, wsId, memberId);
    }

    const [channel] = await tx
      .insert(schema.channels)
      .values({ workspaceId: wsId, type: "group_dm", createdBy: userId, memberCount: uniqueMembers.length })
      .returning();
    if (!channel) throw new Error("channel insert returned no row");

    await tx.insert(schema.channelSeq).values({ channelId: channel.id, lastSeq: 0 });
    await tx.insert(schema.channelMembers).values(
      uniqueMembers.map((memberId) => ({
        channelId: channel.id,
        userId: memberId,
        role: memberId === userId ? "owner" : "member",
      })),
    );

    return channel;
  });
}

/** Hides/reopens a DM only for the caller. History and other participants are
 * untouched, which mirrors Slack's "close conversation" behaviour. */
export async function setDirectMessageClosed(
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  channelId: EntityIdInput,
  closed: boolean,
) {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  const chId = asEntityId(channelId);
  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    await requireWorkspaceMembership(tx, wsId, userId);
    await requireChannelMembership(tx, chId, userId);
    const [channel] = await tx.select().from(schema.channels).where(eq(schema.channels.id, chId)).limit(1);
    if (!channel || channel.workspaceId !== wsId) throw new NotFoundException("direct message not found");
    if (channel.type !== "dm" && channel.type !== "group_dm") {
      throw new BadRequestException("only direct messages can be closed");
    }
    await tx
      .update(schema.channelMembers)
      .set({ isClosed: closed })
      .where(and(eq(schema.channelMembers.channelId, chId), eq(schema.channelMembers.userId, userId)));
    return { closed };
  });
}

/** Promotes an existing group DM into a private channel in place, preserving
 * every message, attachment and participant. */
export async function convertGroupDirectMessage(
  redis: Redis | null,
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  channelId: EntityIdInput,
  name: string,
) {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  const chId = asEntityId(channelId);
  const channel = await withTenant({ workspaceId: wsId, userId }, async (tx) => {
    const workspaceMembership = await requireWorkspaceMembership(tx, wsId, userId);
    const channelMembership = await requireChannelMembership(tx, chId, userId);
    const [current] = await tx.select().from(schema.channels).where(eq(schema.channels.id, chId)).limit(1);
    if (!current || current.workspaceId !== wsId) throw new NotFoundException("group direct message not found");
    if (current.type !== "group_dm") throw new BadRequestException("only a group DM can be converted");
    if (channelMembership.role !== "owner" && workspaceMembership.role !== "owner" && workspaceMembership.role !== "admin") {
      throw new ForbiddenException("only the group owner or a workspace admin can convert this conversation");
    }
    const [duplicate] = await tx
      .select({ id: schema.channels.id })
      .from(schema.channels)
      .where(
        and(
          eq(schema.channels.workspaceId, wsId),
          sql`${schema.channels.id} <> ${chId}`,
          sql`lower(${schema.channels.name}) = lower(${name.trim()})`,
        ),
      )
      .limit(1);
    if (duplicate) throw new BadRequestException("a channel with that name already exists");
    const [updated] = await tx
      .update(schema.channels)
      .set({ type: "private", name: name.trim() })
      .where(eq(schema.channels.id, chId))
      .returning();
    return updated!;
  });

  if (redis) {
    await publishRoomEvent(redis, {
      room: `ch:${chId}`,
      event: { type: "channel:updated", payload: { channel: toWireChannel(channel) } },
    });
  }
  return toWireChannel(channel);
}

/** Standalone membership check (opens its own transaction) — for callers
 * like the gateway's `channel:join` handler that don't already have a `tx`. */
export async function checkChannelMembership(
  workspaceId: EntityIdInput,
  channelId: EntityIdInput,
  userId: EntityIdInput,
): Promise<boolean> {
  const wsId = asEntityId(workspaceId);
  const chId = asEntityId(channelId);
  const uId = asEntityId(userId);
  try {
    await withTenant({ workspaceId: wsId, userId: uId }, (tx) => requireChannelMembership(tx, chId, uId));
    return true;
  } catch {
    return false;
  }
}

export async function checkWorkspaceMembership(workspaceId: EntityIdInput, userId: EntityIdInput): Promise<boolean> {
  const wsId = asEntityId(workspaceId);
  const uId = asEntityId(userId);
  try {
    await withTenant({ workspaceId: wsId, userId: uId }, (tx) => requireWorkspaceMembership(tx, wsId, uId));
    return true;
  } catch {
    return false;
  }
}