import { BadRequestException, ForbiddenException, NotFoundException } from "@nestjs/common";
import { and, asc, desc, eq, gt, inArray, isNull, lt, or, sql } from "drizzle-orm";
import type Redis from "ioredis";
import {
  ForwardedMessageSnapshot as ForwardedMessageSnapshotSchema,
  type ChannelMember,
  type ForwardedMessageSnapshot,
  type ForwardMessageRequest,
  type Message,
  type MessageReactionSummary,
  type ThreadSubscriptionStatus,
  type ThreadSummary,
} from "@slackwsh/contracts";
import { schema, withTenant } from "@slackwsh/data";
import { blocksToPlainText, can, parseMentions, plainTextToBlocksV1, sanitizeMessageBlocks } from "@slackwsh/core";
import { asEntityId, requireChannelMembership, requireWorkspaceMembership, type EntityIdInput } from "./channels";
import { publishRoomEvent } from "./events-bus";
import { enqueueInTx } from "./jobs";

const UNIQUE_VIOLATION = "23505";

export interface SendMessageInput {
  clientMsgId: string;
  text: string;
  blocks?: unknown;
  parentId?: EntityIdInput | null;
  isBroadcast?: boolean;
  /** Server-trusted snapshot. ClientSendMessage strips this field, so only
   * forwardMessage can create embedded forwarded-message cards. */
  forwardedMessage?: ForwardedMessageSnapshot;
}

function attachmentIdsFromBlocks(blocks: unknown): number[] {
  if (!blocks || typeof blocks !== "object") return [];
  const record = blocks as { attachments?: unknown; doc?: { attachments?: unknown } };
  const raw = Array.isArray(record.doc?.attachments)
    ? record.doc.attachments
    : Array.isArray(record.attachments)
      ? record.attachments
      : [];
  return Array.from(new Set(raw
    .map((item) => Number((item as { id?: unknown })?.id))
    .filter((id) => Number.isInteger(id) && id > 0)));
}

function canonicalizeAttachments(
  blocks: unknown,
  rows: Array<typeof schema.attachments.$inferSelect>,
): Record<string, unknown> {
  if (!blocks || typeof blocks !== "object") return plainTextToBlocksV1("") as unknown as Record<string, unknown>;
  if (rows.length === 0) return blocks as Record<string, unknown>;
  const record = blocks as Record<string, unknown>;
  const attachments = rows.map((file) => ({
    id: file.id,
    name: file.originalName,
    type: file.mimeType,
    size: file.size,
    category: file.category,
    url: `/workspaces/${file.workspaceId}/channels/${file.channelId}/files/${file.id}/access`,
  }));
  const doc = record.doc && typeof record.doc === "object"
    ? { ...(record.doc as Record<string, unknown>), attachments }
    : { type: "doc", content: [], attachments };
  return { ...record, attachments, doc };
}

function removeUnavailableAttachments(blocks: unknown, availableIds: Set<number>) {
  if (!blocks || typeof blocks !== "object") return blocks;
  const record = blocks as Record<string, unknown>;
  const filter = (value: unknown) => Array.isArray(value)
    ? value.filter((item) => {
        const id = Number((item as { id?: unknown })?.id);
        // Legacy local uploads used UUIDs and have no attachment DB row; keep
        // them so historical messages remain downloadable.
        return !Number.isInteger(id) || availableIds.has(id);
      })
    : value;
  const doc = record.doc && typeof record.doc === "object"
    ? { ...(record.doc as Record<string, unknown>), attachments: filter((record.doc as Record<string, unknown>).attachments) }
    : record.doc;
  return { ...record, attachments: filter(record.attachments), doc };
}

/**
 * The write path from ARCHITECTURE §4.3: allocate the per-channel seq
 * (ADR-001) and insert inside one transaction, then a best-effort broadcast
 * after commit. Idempotent by construction (I3) via `UNIQUE(channel_id,
 * client_msg_id)` — a retry that hits the constraint returns the row that
 * already won, rather than erroring the client's retry loop.
 */
export async function sendMessage(
  redis: Redis,
  workspaceId: EntityIdInput,
  channelId: EntityIdInput,
  actorUserId: EntityIdInput,
  input: SendMessageInput,
) {
  const wsId = asEntityId(workspaceId);
  const chId = asEntityId(channelId);
  const userId = asEntityId(actorUserId);
  const parentId = input.parentId != null ? asEntityId(input.parentId) : null;
  let message: typeof schema.messages.$inferSelect;
  let updatedThreadRoot: typeof schema.messages.$inferSelect | null = null;

  try {
    message = await withTenant({ workspaceId: wsId, userId }, async (tx) => {
      const membership = await requireChannelMembership(tx, chId, userId);
      const workspaceMembership = await requireWorkspaceMembership(tx, wsId, userId);
      if (!can({ userId, role: workspaceMembership.role }, "channel:post")) {
        throw new ForbiddenException("insufficient role to post in this channel");
      }
      void membership;

      const [seqRow] = await tx
        .update(schema.channelSeq)
        .set({ lastSeq: sql`${schema.channelSeq.lastSeq} + 1` })
        .where(eq(schema.channelSeq.channelId, chId))
        .returning({ lastSeq: schema.channelSeq.lastSeq });
      if (!seqRow) throw new NotFoundException("channel has no sequence counter");

      const sanitizedBlocks = sanitizeMessageBlocks(input.blocks, input.text);
      let blocks: Record<string, unknown> = input.forwardedMessage
        ? { ...sanitizedBlocks, forwardedMessage: input.forwardedMessage }
        : sanitizedBlocks;
      const text = input.blocks == null ? input.text : blocksToPlainText(sanitizedBlocks);
      const attachmentIds = attachmentIdsFromBlocks(blocks);
      let attachmentRows: Array<typeof schema.attachments.$inferSelect> = [];
      if (!text.trim() && attachmentIds.length === 0 && !input.forwardedMessage) {
        throw new BadRequestException("message text or at least one file is required");
      }
      if (attachmentIds.length > 10) throw new BadRequestException("a message can contain at most 10 files");
      if (attachmentIds.length > 0) {
        attachmentRows = await tx.select().from(schema.attachments).where(and(
          inArray(schema.attachments.id, attachmentIds),
          eq(schema.attachments.workspaceId, wsId),
          eq(schema.attachments.channelId, chId),
          eq(schema.attachments.uploadedBy, userId),
          eq(schema.attachments.status, "ready"),
          isNull(schema.attachments.messageId),
          isNull(schema.attachments.deletedAt),
        ));
        if (attachmentRows.length !== attachmentIds.length) {
          throw new BadRequestException("one or more files are invalid, incomplete, already used, or belong to another channel");
        }
        const byId = new Map(attachmentRows.map((row) => [row.id, row]));
        blocks = canonicalizeAttachments(blocks, attachmentIds.map((id) => byId.get(id)!));
      }
      const [inserted] = await tx
        .insert(schema.messages)
        .values({
          workspaceId: wsId,
          channelId: chId,
          seq: seqRow.lastSeq,
          clientMsgId: input.clientMsgId,
          authorId: userId,
          type: "text",
          text,
          blocks,
          parentId,
          isBroadcast: input.isBroadcast ?? false,
        })
        .returning();
      if (!inserted) throw new Error("message insert returned no row");

      if (attachmentIds.length > 0) {
        await tx.update(schema.attachments).set({ messageId: inserted.id }).where(inArray(schema.attachments.id, attachmentIds));
      }

      const parsed = parseMentions(text);
      const broadMentions = parsed.filter((m) => m.kind !== "user");
      const userMentions = parsed.filter((m): m is Extract<(typeof parsed)[number], { kind: "user" }> => m.kind === "user");
      const channelMemberRows = await tx
        .select({ userId: schema.channelMembers.userId })
        .from(schema.channelMembers)
        .where(eq(schema.channelMembers.channelId, chId));
      const channelMemberIds = new Set(channelMemberRows.map((row) => row.userId));
      const mentionRecipientIds = new Set<number>();
      const directlyMentionedUserIds = new Set<number>();

      const mentionRows: Array<{ messageId: number; targetType: string; targetId: number | null }> = broadMentions.map((m) => ({
        messageId: inserted.id,
        targetType: m.kind,
        targetId: null,
      }));
      if (broadMentions.length > 0) {
        for (const memberId of channelMemberIds) {
          if (memberId !== userId) mentionRecipientIds.add(memberId);
        }
      }

      if (userMentions.length > 0) {
        // Only resolves to users who are actual members of this workspace —
        // a handle match outside the workspace shouldn't leak that the
        // account exists, and mentioning a non-member wouldn't notify them.
        const handles = Array.from(new Set(userMentions.map((m) => m.handle)));
        const matches = await tx
          .select({ id: schema.users.id })
          .from(schema.users)
          .innerJoin(schema.workspaceMembers, eq(schema.workspaceMembers.userId, schema.users.id))
          .where(and(inArray(schema.users.username, handles), eq(schema.workspaceMembers.workspaceId, wsId)));
        for (const match of matches) {
          mentionRows.push({ messageId: inserted.id, targetType: "user", targetId: match.id });
          directlyMentionedUserIds.add(match.id);
          if (match.id !== userId && channelMemberIds.has(match.id)) mentionRecipientIds.add(match.id);
        }
      }

      if (mentionRows.length > 0) {
        await tx.insert(schema.messageMentions).values(mentionRows);
      }
      await enqueueInTx(tx, "notifications", {
        messageId: inserted.id,
        workspaceId: wsId,
        channelId: chId,
        authorId: userId,
      });
      await enqueueInTx(tx, "search_index", { messageId: inserted.id, workspaceId: wsId });
      const mentionRecipients = [...mentionRecipientIds];
      if (mentionRecipients.length > 0) {
        await tx
          .update(schema.channelMembers)
          .set({ mentionCount: sql`${schema.channelMembers.mentionCount} + 1` })
          .where(and(eq(schema.channelMembers.channelId, chId), inArray(schema.channelMembers.userId, mentionRecipients)));
      }

      if (parentId == null) {
        await tx.update(schema.channels).set({ lastMessageAt: new Date() }).where(eq(schema.channels.id, chId));
      } else {
        const [rootAfterReply] = await tx
          .update(schema.messages)
          .set({
            threadReplyCount: sql`${schema.messages.threadReplyCount} + 1`,
            threadLastReplyAt: new Date(),
            revision: sql`${schema.messages.revision} + 1`,
          })
          .where(and(
            eq(schema.messages.id, parentId),
            eq(schema.messages.channelId, chId),
            isNull(schema.messages.parentId),
            isNull(schema.messages.deletedAt),
          ))
          .returning();
        if (!rootAfterReply) throw new BadRequestException("thread root was not found in this channel");
        updatedThreadRoot = rootAfterReply;

        // Participating in a thread follows it and marks the sender's own
        // reply read. Root authors and directly-mentioned members are
        // auto-followed only if they have not explicitly unfollowed before.
        await tx
          .insert(schema.threadSubscriptions)
          .values({
            userId,
            rootMessageId: parentId,
            channelId: chId,
            workspaceId: wsId,
            lastReadReplySeq: inserted.seq,
            reason: "reply",
            isMuted: false,
          })
          .onConflictDoUpdate({
            target: [schema.threadSubscriptions.userId, schema.threadSubscriptions.rootMessageId],
            set: {
              lastReadReplySeq: sql`greatest(${schema.threadSubscriptions.lastReadReplySeq}, ${inserted.seq})`,
              reason: "reply",
              isMuted: false,
            },
          });

        const passiveFollowerIds = new Set<number>([rootAfterReply.authorId, ...directlyMentionedUserIds]);
        passiveFollowerIds.delete(userId);
        const passiveFollowers = [...passiveFollowerIds].filter((id) => channelMemberIds.has(id));
        if (passiveFollowers.length > 0) {
          await tx
            .insert(schema.threadSubscriptions)
            .values(passiveFollowers.map((followerId) => ({
              userId: followerId,
              rootMessageId: parentId,
              channelId: chId,
              workspaceId: wsId,
              lastReadReplySeq: 0,
              reason: followerId === rootAfterReply.authorId ? "thread_author" : "mention",
              isMuted: false,
            })))
            .onConflictDoNothing();
        }
      }

      return inserted;
    });
  } catch (err: any) {
    // A unique-violation on retry (I3) must be resolved in a *fresh*
    // transaction, not recovered inside the failed one: postgres.js (the
    // driver behind libs/data) still fails the whole `sql.begin()` block at
    // commit even after a raw `ROLLBACK TO SAVEPOINT` inside the callback —
    // unlike node-postgres, it doesn't support continuing a transaction
    // past an error via manually-issued savepoint SQL. (This looks like the
    // Slacknew/Elixir project's transaction-poisoning issue and shares its
    // root cause — Postgres aborting a transaction after any error — but
    // the *fix* differs because the driver differs: no in-transaction
    // recovery is available here, so we let it roll back and re-read after.)
    if (err?.code !== UNIQUE_VIOLATION) throw err;

    // A plain getDb() query here (no tenant context set) would return zero
    // rows regardless — messages' RLS policy is workspace-scoped only, with
    // no self_membership escape hatch like workspaces/workspace_members
    // have. Needs its own withTenant, same as the original attempt.
    const existing = await withTenant({ workspaceId: wsId, userId }, (tx) =>
      tx
        .select()
        .from(schema.messages)
        .where(and(eq(schema.messages.channelId, chId), eq(schema.messages.clientMsgId, input.clientMsgId)))
        .limit(1),
    );
    if (!existing[0]) throw err;
    message = existing[0];

    // The original request committed successfully but its HTTP response was
    // lost. Re-read the already-updated root so the retry can still broadcast
    // an authoritative thread summary instead of incrementing it a second
    // time or leaving the sender's UI stale.
    if (message.parentId != null) {
      const [root] = await withTenant({ workspaceId: wsId, userId }, (tx) =>
        tx
          .select()
          .from(schema.messages)
          .where(and(eq(schema.messages.id, message.parentId!), eq(schema.messages.channelId, chId)))
          .limit(1),
      );
      updatedThreadRoot = root ?? null;
    }
  }

  const wireMessage = toWireMessage(message);
  await publishRoomEvent(redis, {
    room: `ch:${chId}`,
    event: { type: "message:created", payload: { message: wireMessage } },
  });
  if (updatedThreadRoot) {
    // Reply metadata lives on the root message. Broadcasting that updated row
    // makes the main timeline's "N replies / Last reply" summary converge in
    // real time on both the sender and every other client already in the room.
    await publishRoomEvent(redis, {
      room: `ch:${chId}`,
      event: { type: "message:edited", payload: { message: toWireMessage(updatedThreadRoot) } },
    });
  }

  // Personal fanout (§4.3 u:{userId}): members who are not currently joined
  // to this channel's socket room (typical for DMs — the other person is
  // elsewhere in the app) still need a live event. Same payload shape as
  // the channel room; clients dedupe by message id if they receive both.
  const peerIds = await withTenant({ workspaceId: wsId, userId }, async (tx) => {
    if (message.parentId != null) {
      const rows = await tx
        .select({ userId: schema.threadSubscriptions.userId })
        .from(schema.threadSubscriptions)
        .where(and(
          eq(schema.threadSubscriptions.rootMessageId, message.parentId),
          eq(schema.threadSubscriptions.isMuted, false),
        ));
      // Include the sender for same-account tabs/devices. The receiving client
      // suppresses its own notification while still refreshing thread state.
      return rows.map((row) => row.userId);
    }
    const rows = await tx
      .select({ userId: schema.channelMembers.userId })
      .from(schema.channelMembers)
      .where(eq(schema.channelMembers.channelId, chId));
    return rows.map((row) => row.userId).filter((id) => id !== userId);
  });
  for (const peerId of peerIds) {
    await publishRoomEvent(redis, {
      room: `u:${peerId}`,
      event: {
        type: "message:created",
        payload: { message: wireMessage, workspaceId: wsId, channelId: chId },
      },
    });
    if (message.parentId != null) {
      const status = await withTenant({ workspaceId: wsId, userId }, (tx) =>
        threadSubscriptionStatusInTx(tx, wsId, chId, peerId, message.parentId!),
      );
      await publishThreadSubscriptionStatus(redis, peerId, status);
    }
  }

  return message;
}

/** Slack-style forwarding: validate access to the source once, snapshot the
 * original author/context, then post a trusted preview into each destination.
 * Destination membership and posting permissions are enforced independently
 * by sendMessage. */
export async function forwardMessage(
  redis: Redis,
  workspaceId: EntityIdInput,
  sourceChannelId: EntityIdInput,
  actorUserId: EntityIdInput,
  sourceMessageId: EntityIdInput,
  input: ForwardMessageRequest,
) {
  const wsId = asEntityId(workspaceId);
  const sourceChId = asEntityId(sourceChannelId);
  const userId = asEntityId(actorUserId);
  const sourceMsgId = asEntityId(sourceMessageId);
  const snapshot = await withTenant({ workspaceId: wsId, userId }, async (tx) => {
    await requireChannelMembership(tx, sourceChId, userId);
    await requireWorkspaceMembership(tx, wsId, userId);
    // Validate every destination before the first insert so an inaccessible
    // room cannot cause an avoidable half-forward across a multi-select.
    for (const destination of input.destinations) {
      await requireChannelMembership(tx, destination.channelId, userId);
    }
    const source = await requireMessageInChannel(tx, sourceMsgId, sourceChId);
    if (source.workspaceId !== wsId || source.deletedAt) throw new NotFoundException("message not found");

    const existingSnapshot = ForwardedMessageSnapshotSchema.safeParse(
      (source.blocks as Record<string, unknown> | null)?.forwardedMessage,
    );
    if (existingSnapshot.success && !source.text.trim()) return existingSnapshot.data;

    const [[author], [channel]] = await Promise.all([
      tx.select({ id: schema.users.id, name: schema.users.name, avatarUrl: schema.users.avatarUrl })
        .from(schema.users).where(eq(schema.users.id, source.authorId)).limit(1),
      tx.select({ id: schema.channels.id, name: schema.channels.name, type: schema.channels.type })
        .from(schema.channels).where(eq(schema.channels.id, sourceChId)).limit(1),
    ]);
    if (!author || !channel) throw new NotFoundException("message context not found");

    const attachmentIds = attachmentIdsFromBlocks(source.blocks);
    const files = attachmentIds.length > 0
      ? await tx.select({ name: schema.attachments.originalName }).from(schema.attachments).where(and(
          inArray(schema.attachments.id, attachmentIds),
          eq(schema.attachments.status, "ready"),
          isNull(schema.attachments.deletedAt),
        ))
      : [];
    return ForwardedMessageSnapshotSchema.parse({
      messageId: source.id,
      channelId: source.channelId,
      parentId: source.parentId,
      channelName: channel.name,
      channelType: channel.type,
      authorId: author.id,
      authorName: author.name,
      authorAvatarUrl: author.avatarUrl,
      text: source.text,
      createdAt: source.createdAt.toISOString(),
      attachmentNames: files.map((file) => file.name),
    });
  });

  const note = input.note.trim();
  const blocks = plainTextToBlocksV1(note);
  const messages = [];
  for (const destination of input.destinations) {
    messages.push(await sendMessage(redis, wsId, destination.channelId, userId, {
      clientMsgId: destination.clientMsgId,
      text: note,
      blocks,
      forwardedMessage: snapshot,
    }));
  }
  return messages;
}

/** Fetches the message and asserts it belongs to `channelId` before any
 * write — the cross-resource-id check every caller-supplied-id-pair action
 * needs (RLS alone only scopes by workspace_id, not this relationship). */
async function requireMessageInChannel(tx: any, messageId: EntityIdInput, channelId: EntityIdInput) {
  const msgId = asEntityId(messageId);
  const chId = asEntityId(channelId);
  const [message] = await tx.select().from(schema.messages).where(eq(schema.messages.id, msgId)).limit(1);
  if (!message || message.channelId !== chId) throw new NotFoundException("message not found in this channel");
  return message;
}

export async function editMessage(
  redis: Redis,
  workspaceId: EntityIdInput,
  channelId: EntityIdInput,
  actorUserId: EntityIdInput,
  messageId: EntityIdInput,
  input: { text: string; blocks?: unknown },
) {
  const wsId = asEntityId(workspaceId);
  const chId = asEntityId(channelId);
  const userId = asEntityId(actorUserId);
  const msgId = asEntityId(messageId);
  const blocks = sanitizeMessageBlocks(input.blocks, input.text);
  const text = input.blocks == null ? input.text : blocksToPlainText(blocks);
  if (!text.trim()) throw new BadRequestException("message text is required");
  const message = await withTenant({ workspaceId: wsId, userId }, async (tx) => {
    const workspaceMembership = await requireWorkspaceMembership(tx, wsId, userId);
    const existing = await requireMessageInChannel(tx, msgId, chId);
    const forwarded = ForwardedMessageSnapshotSchema.safeParse(
      (existing.blocks as Record<string, unknown> | null)?.forwardedMessage,
    );
    const nextBlocks = forwarded.success ? { ...blocks, forwardedMessage: forwarded.data } : blocks;

    const isOwn = existing.authorId === userId;
    const action = isOwn ? "message:edit_own" : "message:edit_any";
    if (!can({ userId, role: workspaceMembership.role }, action, { isOwnResource: isOwn })) {
      throw new ForbiddenException("insufficient permission to edit this message");
    }

    const [updated] = await tx
      .update(schema.messages)
      .set({
        text,
        blocks: nextBlocks,
        revision: sql`${schema.messages.revision} + 1`,
        editedAt: new Date(),
      })
      .where(eq(schema.messages.id, msgId))
      .returning();
    return updated!;
  });

  await publishRoomEvent(redis, {
    room: `ch:${chId}`,
    event: { type: "message:edited", payload: { message: toWireMessage(message) } },
  });
  return message;
}

export async function deleteMessage(
  redis: Redis,
  workspaceId: EntityIdInput,
  channelId: EntityIdInput,
  actorUserId: EntityIdInput,
  messageId: EntityIdInput,
) {
  const wsId = asEntityId(workspaceId);
  const chId = asEntityId(channelId);
  const userId = asEntityId(actorUserId);
  const msgId = asEntityId(messageId);
  const { seq, updatedRoot } = await withTenant({ workspaceId: wsId, userId }, async (tx) => {
    const workspaceMembership = await requireWorkspaceMembership(tx, wsId, userId);
    const existing = await requireMessageInChannel(tx, msgId, chId);

    const isOwn = existing.authorId === userId;
    const action = isOwn ? "message:delete_own" : "message:delete_any";
    if (!can({ userId, role: workspaceMembership.role }, action, { isOwnResource: isOwn })) {
      throw new ForbiddenException("insufficient permission to delete this message");
    }

    await tx
      .update(schema.messages)
      .set({ deletedAt: new Date(), revision: sql`${schema.messages.revision} + 1` })
      .where(eq(schema.messages.id, msgId));

    let updatedRoot: typeof schema.messages.$inferSelect | null = null;
    if (existing.parentId != null) {
      const [summary] = await tx
        .select({
          count: sql<number>`count(*)::int`,
          lastReplyAt: sql<Date | null>`max(${schema.messages.createdAt})`,
        })
        .from(schema.messages)
        .where(and(eq(schema.messages.parentId, existing.parentId), isNull(schema.messages.deletedAt)));

      const [rootAfterReplyDelete] = await tx
        .update(schema.messages)
        .set({
          threadReplyCount: summary?.count ?? 0,
          threadLastReplyAt: summary?.lastReplyAt ?? null,
          revision: sql`${schema.messages.revision} + 1`,
        })
        .where(eq(schema.messages.id, existing.parentId))
        .returning();
      updatedRoot = rootAfterReplyDelete ?? null;
    }

    return { seq: existing.seq, updatedRoot };
  });

  await publishRoomEvent(redis, {
    room: `ch:${chId}`,
    event: { type: "message:deleted", payload: { channelId: chId, messageId: msgId, seq } },
  });
  if (updatedRoot) {
    await publishRoomEvent(redis, {
      room: `ch:${chId}`,
      event: { type: "message:edited", payload: { message: toWireMessage(updatedRoot) } },
    });
  }
}

export async function setReaction(
  redis: Redis,
  workspaceId: EntityIdInput,
  channelId: EntityIdInput,
  actorUserId: EntityIdInput,
  messageId: EntityIdInput,
  emoji: string,
  op: "add" | "remove",
) {
  const wsId = asEntityId(workspaceId);
  const chId = asEntityId(channelId);
  const userId = asEntityId(actorUserId);
  const msgId = asEntityId(messageId);
  await withTenant({ workspaceId: wsId, userId }, async (tx) => {
    await requireWorkspaceMembership(tx, wsId, userId);
    await requireMessageInChannel(tx, msgId, chId);

    if (op === "add") {
      await tx.insert(schema.reactions).values({ messageId: msgId, userId, emoji }).onConflictDoNothing();
    } else {
      await tx
        .delete(schema.reactions)
        .where(and(eq(schema.reactions.messageId, msgId), eq(schema.reactions.userId, userId), eq(schema.reactions.emoji, emoji)));
    }
  });

  await publishRoomEvent(redis, {
    room: `ch:${chId}`,
    event: { type: "reaction:changed", payload: { reaction: { messageId: msgId, userId, emoji }, op } },
  });
}

export async function setPin(
  redis: Redis,
  workspaceId: EntityIdInput,
  channelId: EntityIdInput,
  actorUserId: EntityIdInput,
  messageId: EntityIdInput,
  op: "add" | "remove",
) {
  const wsId = asEntityId(workspaceId);
  const chId = asEntityId(channelId);
  const userId = asEntityId(actorUserId);
  const msgId = asEntityId(messageId);
  await withTenant({ workspaceId: wsId, userId }, async (tx) => {
    const workspaceMembership = await requireWorkspaceMembership(tx, wsId, userId);
    if (!can({ userId, role: workspaceMembership.role }, "message:pin")) {
      throw new ForbiddenException("insufficient role to pin messages");
    }
    // The exact gap this guards against (see project memory): a member could
    // otherwise pin a message from a *different* channel into this one by
    // supplying a foreign messageId — pins' own RLS only constrains
    // channel_id/workspace_id, both legitimately the caller's own.
    await requireMessageInChannel(tx, msgId, chId);

    if (op === "add") {
      await tx.insert(schema.pins).values({ channelId: chId, messageId: msgId, pinnedBy: userId }).onConflictDoNothing();
    } else {
      await tx.delete(schema.pins).where(and(eq(schema.pins.channelId, chId), eq(schema.pins.messageId, msgId)));
    }
  });

  await publishRoomEvent(redis, {
    room: `ch:${chId}`,
    event: { type: "pin:changed", payload: { channelId: chId, messageId: msgId, op } },
  });
}

export async function markChannelRead(
  redis: Redis,
  workspaceId: EntityIdInput,
  channelId: EntityIdInput,
  actorUserId: EntityIdInput,
  seq: number,
) {
  const wsId = asEntityId(workspaceId);
  const chId = asEntityId(channelId);
  const userId = asEntityId(actorUserId);
  const channelMember = await withTenant({ workspaceId: wsId, userId }, async (tx) => {
    await requireChannelMembership(tx, chId, userId);
    // I4: never decreases except by explicit user action — GREATEST enforces
    // that even under out-of-order acks (e.g. a stale client catching up).
    const [updated] = await tx
      .update(schema.channelMembers)
      .set({
        lastReadSeq: sql`greatest(${schema.channelMembers.lastReadSeq}, ${seq})`,
        lastReadAt: new Date(),
        mentionCount: 0,
      })
      .where(and(eq(schema.channelMembers.channelId, chId), eq(schema.channelMembers.userId, userId)))
      .returning();
    return updated!;
  });

  await publishRoomEvent(redis, {
    room: `u:${userId}`,
    event: { type: "read:updated", payload: { channelMember: toWireChannelMember(channelMember) } },
  });
}

async function threadSubscriptionStatusInTx(
  tx: any,
  workspaceId: number,
  channelId: number,
  userId: number,
  rootMessageId: number,
): Promise<ThreadSubscriptionStatus> {
  const [subscription] = await tx
    .select()
    .from(schema.threadSubscriptions)
    .where(and(
      eq(schema.threadSubscriptions.userId, userId),
      eq(schema.threadSubscriptions.rootMessageId, rootMessageId),
    ))
    .limit(1);
  const lastReadReplySeq = subscription?.lastReadReplySeq ?? 0;
  const [unread] = await tx
    .select({ count: sql<number>`count(*)::int` })
    .from(schema.messages)
    .where(and(
      eq(schema.messages.parentId, rootMessageId),
      gt(schema.messages.seq, lastReadReplySeq),
      isNull(schema.messages.deletedAt),
      sql`${schema.messages.authorId} <> ${userId}`,
    ));
  return {
    workspaceId,
    channelId,
    rootMessageId,
    following: Boolean(subscription && !subscription.isMuted),
    reason: subscription?.reason ?? null,
    lastReadReplySeq,
    unreadReplyCount: Number(unread?.count ?? 0),
  };
}

async function publishThreadSubscriptionStatus(redis: Redis, userId: number, status: ThreadSubscriptionStatus) {
  await publishRoomEvent(redis, {
    room: `u:${userId}`,
    event: { type: "thread:subscription-updated", payload: { status } },
  });
}

export async function getThreadSubscriptionStatus(
  workspaceId: EntityIdInput,
  channelId: EntityIdInput,
  actorUserId: EntityIdInput,
  rootMessageId: EntityIdInput,
) {
  const wsId = asEntityId(workspaceId);
  const chId = asEntityId(channelId);
  const userId = asEntityId(actorUserId);
  const rootId = asEntityId(rootMessageId);
  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    await requireChannelMembership(tx, chId, userId);
    const root = await requireMessageInChannel(tx, rootId, chId);
    if (root.parentId !== null) throw new BadRequestException("rootMessageId must be a top-level message");
    return threadSubscriptionStatusInTx(tx, wsId, chId, userId, rootId);
  });
}

export async function setThreadFollowing(
  redis: Redis,
  workspaceId: EntityIdInput,
  channelId: EntityIdInput,
  actorUserId: EntityIdInput,
  rootMessageId: EntityIdInput,
  following: boolean,
) {
  const wsId = asEntityId(workspaceId);
  const chId = asEntityId(channelId);
  const userId = asEntityId(actorUserId);
  const rootId = asEntityId(rootMessageId);
  const status = await withTenant({ workspaceId: wsId, userId }, async (tx) => {
    await requireChannelMembership(tx, chId, userId);
    const root = await requireMessageInChannel(tx, rootId, chId);
    if (root.parentId !== null) throw new BadRequestException("rootMessageId must be a top-level message");
    const [latest] = await tx
      .select({ seq: schema.messages.seq })
      .from(schema.messages)
      .where(and(eq(schema.messages.parentId, rootId), isNull(schema.messages.deletedAt)))
      .orderBy(desc(schema.messages.seq))
      .limit(1);

    await tx
      .insert(schema.threadSubscriptions)
      .values({
        userId,
        rootMessageId: rootId,
        channelId: chId,
        workspaceId: wsId,
        lastReadReplySeq: following ? (latest?.seq ?? 0) : 0,
        reason: following ? "manual" : "manual_unfollow",
        isMuted: !following,
      })
      .onConflictDoUpdate({
        target: [schema.threadSubscriptions.userId, schema.threadSubscriptions.rootMessageId],
        set: following
          ? {
              isMuted: false,
              reason: "manual",
              lastReadReplySeq: sql`greatest(${schema.threadSubscriptions.lastReadReplySeq}, ${latest?.seq ?? 0})`,
            }
          : { isMuted: true, reason: "manual_unfollow" },
      });
    return threadSubscriptionStatusInTx(tx, wsId, chId, userId, rootId);
  });
  await publishThreadSubscriptionStatus(redis, userId, status);
  return status;
}

export async function markThreadRead(
  redis: Redis,
  workspaceId: EntityIdInput,
  channelId: EntityIdInput,
  actorUserId: EntityIdInput,
  rootMessageId: EntityIdInput,
  seq: number,
) {
  const wsId = asEntityId(workspaceId);
  const chId = asEntityId(channelId);
  const userId = asEntityId(actorUserId);
  const rootId = asEntityId(rootMessageId);
  const status = await withTenant({ workspaceId: wsId, userId }, async (tx) => {
    await requireChannelMembership(tx, chId, userId);
    const root = await requireMessageInChannel(tx, rootId, chId);
    if (root.parentId !== null) throw new BadRequestException("rootMessageId must be a top-level message");

    const existing = await tx
      .select()
      .from(schema.threadSubscriptions)
      .where(and(eq(schema.threadSubscriptions.userId, userId), eq(schema.threadSubscriptions.rootMessageId, rootId)))
      .limit(1);

    if (existing.length === 0) {
      await tx.insert(schema.threadSubscriptions).values({
        userId,
        rootMessageId: rootId,
        channelId: chId,
        workspaceId: wsId,
        lastReadReplySeq: seq,
        reason: "viewed",
        isMuted: true,
      });
    } else {
      await tx
        .update(schema.threadSubscriptions)
        .set({ lastReadReplySeq: sql`greatest(${schema.threadSubscriptions.lastReadReplySeq}, ${seq})` })
        .where(and(eq(schema.threadSubscriptions.userId, userId), eq(schema.threadSubscriptions.rootMessageId, rootId)));
    }
    return threadSubscriptionStatusInTx(tx, wsId, chId, userId, rootId);
  });
  await publishThreadSubscriptionStatus(redis, userId, status);
  return status;
}

export async function listFollowedThreads(
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  limit = 50,
): Promise<ThreadSummary[]> {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    await requireWorkspaceMembership(tx, wsId, userId);
    const followed = await tx
      .select({
        subscription: schema.threadSubscriptions,
        rootMessage: schema.messages,
        channelName: schema.channels.name,
      })
      .from(schema.threadSubscriptions)
      .innerJoin(schema.messages, eq(schema.messages.id, schema.threadSubscriptions.rootMessageId))
      .innerJoin(schema.channels, eq(schema.channels.id, schema.threadSubscriptions.channelId))
      .innerJoin(
        schema.channelMembers,
        and(
          eq(schema.channelMembers.channelId, schema.threadSubscriptions.channelId),
          eq(schema.channelMembers.userId, userId),
        ),
      )
      .where(and(
        eq(schema.threadSubscriptions.workspaceId, wsId),
        eq(schema.threadSubscriptions.userId, userId),
        eq(schema.threadSubscriptions.isMuted, false),
        isNull(schema.messages.deletedAt),
      ))
      .orderBy(sql`${schema.messages.threadLastReplyAt} desc nulls last`, desc(schema.messages.createdAt))
      .limit(Math.max(1, Math.min(limit, 100)));
    if (followed.length === 0) return [];

    const rootIds = followed.map((row) => row.rootMessage.id);
    const replyRows = await tx
      .select({ message: schema.messages })
      .from(schema.messages)
      .where(and(inArray(schema.messages.parentId, rootIds), isNull(schema.messages.deletedAt)))
      .orderBy(asc(schema.messages.seq));
    const allAuthorIds = [...new Set([
      ...followed.map((row) => row.rootMessage.authorId),
      ...replyRows.map((row) => row.message.authorId),
    ])];
    const people = await tx
      .select({ userId: schema.users.id, name: schema.users.name, avatarUrl: schema.users.avatarUrl })
      .from(schema.users)
      .where(inArray(schema.users.id, allAuthorIds));
    const personById = new Map(people.map((person) => [person.userId, person]));

    return followed
      .map((row) => {
        const replies = replyRows
          .map((reply) => reply.message)
          .filter((reply) => reply.parentId === row.rootMessage.id);
        const latestReply = replies.at(-1) ?? null;
        const participantIds = [...new Set([row.rootMessage.authorId, ...replies.map((reply) => reply.authorId)])];
        return {
          rootMessage: toWireMessage(row.rootMessage),
          latestReply: latestReply ? toWireMessage(latestReply) : null,
          channelName: row.channelName,
          following: true,
          reason: row.subscription.reason,
          lastReadReplySeq: row.subscription.lastReadReplySeq,
          unreadReplyCount: replies.filter(
            (reply) => reply.seq > row.subscription.lastReadReplySeq && reply.authorId !== userId,
          ).length,
          participants: participantIds
            .map((id) => personById.get(id))
            .filter((person): person is NonNullable<typeof person> => Boolean(person)),
        } satisfies ThreadSummary;
      })
      .sort((left, right) => {
        const leftAt = left.latestReply?.createdAt ?? left.rootMessage.createdAt;
        const rightAt = right.latestReply?.createdAt ?? right.rootMessage.createdAt;
        return new Date(rightAt).getTime() - new Date(leftAt).getTime();
      });
  });
}

export interface ScrollbackOptions {
  afterSeq?: number;
  beforeSeq?: number;
  limit: number;
}

/** Channel scrollback (§5.3 catch-up mechanism): top-level messages only,
 * i.e. `parent_id IS NULL OR is_broadcast` — thread replies live in the
 * thread view (getThread), per I5's channel/thread split (ADR-014). */
/** Resolve one permalink target without forcing the client to page through
 * all older history. Access and deletion rules match the ordinary read path. */
export async function getMessage(
  workspaceId: EntityIdInput,
  channelId: EntityIdInput,
  actorUserId: EntityIdInput,
  messageId: EntityIdInput,
): Promise<Message> {
  const wsId = asEntityId(workspaceId);
  const chId = asEntityId(channelId);
  const userId = asEntityId(actorUserId);
  const msgId = asEntityId(messageId);
  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    await requireChannelMembership(tx, chId, userId);
    const message = await requireMessageInChannel(tx, msgId, chId);
    if (message.workspaceId !== wsId) throw new NotFoundException("message not found");
    const [wire] = await withReactionSummaries(tx, [scrubDeleted(message)], userId);
    if (!wire) throw new NotFoundException("message not found");
    return wire;
  });
}

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

    const conditions = [eq(schema.messages.channelId, chId), or(isNull(schema.messages.parentId), eq(schema.messages.isBroadcast, true))];
    if (opts.afterSeq !== undefined) conditions.push(gt(schema.messages.seq, opts.afterSeq));
    if (opts.beforeSeq !== undefined) conditions.push(lt(schema.messages.seq, opts.beforeSeq));

    // A request without a cursor opens a conversation, so return the most
    // recent page. `beforeSeq` also reads backwards. Both are reversed back
    // to chronological order before leaving the service; `afterSeq` remains
    // the forward catch-up/gap-repair path.
    const readsBackward = opts.beforeSeq !== undefined || opts.afterSeq === undefined;
    const rows = await tx
      .select()
      .from(schema.messages)
      .where(and(...conditions))
      .orderBy(readsBackward ? desc(schema.messages.seq) : asc(schema.messages.seq))
      .limit(opts.limit);

    return withReactionSummaries(tx, (readsBackward ? rows.reverse() : rows).map(scrubDeleted), userId);
  });
}

export async function getThread(
  workspaceId: EntityIdInput,
  channelId: EntityIdInput,
  actorUserId: EntityIdInput,
  rootMessageId: EntityIdInput,
) {
  const wsId = asEntityId(workspaceId);
  const chId = asEntityId(channelId);
  const userId = asEntityId(actorUserId);
  const rootId = asEntityId(rootMessageId);
  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    await requireChannelMembership(tx, chId, userId);
    await requireMessageInChannel(tx, rootId, chId);
    const rows = await tx
      .select()
      .from(schema.messages)
      .where(and(eq(schema.messages.channelId, chId), eq(schema.messages.parentId, rootId)))
      .orderBy(asc(schema.messages.seq));
    return withReactionSummaries(tx, rows.map(scrubDeleted), userId);
  });
}

/** Explicit user action allowed to move I4's read cursor backwards. */
export async function markMessageUnread(
  redis: Redis,
  workspaceId: EntityIdInput,
  channelId: EntityIdInput,
  actorUserId: EntityIdInput,
  messageId: EntityIdInput,
) {
  const wsId = asEntityId(workspaceId);
  const chId = asEntityId(channelId);
  const userId = asEntityId(actorUserId);
  const msgId = asEntityId(messageId);
  const result = await withTenant({ workspaceId: wsId, userId }, async (tx) => {
    await requireChannelMembership(tx, chId, userId);
    const message = await requireMessageInChannel(tx, msgId, chId);
    if (message.workspaceId !== wsId) throw new NotFoundException("message not found");
    if (message.parentId !== null && !message.isBroadcast) {
      throw new BadRequestException("mark a top-level message unread from the channel view");
    }
    const targetSeq = Math.max(0, message.seq - 1);
    const [updated] = await tx
      .update(schema.channelMembers)
      .set({
        lastReadSeq: sql`least(${schema.channelMembers.lastReadSeq}, ${targetSeq})`,
        lastReadAt: new Date(),
      })
      .where(and(eq(schema.channelMembers.channelId, chId), eq(schema.channelMembers.userId, userId)))
      .returning();
    const [countRow] = await tx
      .select({ count: sql<number>`count(*)::int` })
      .from(schema.messages)
      .where(
        and(
          eq(schema.messages.channelId, chId),
          gt(schema.messages.seq, updated!.lastReadSeq),
          or(isNull(schema.messages.parentId), eq(schema.messages.isBroadcast, true)),
          isNull(schema.messages.deletedAt),
          sql`${schema.messages.authorId} <> ${userId}`,
        ),
      );
    return { channelMember: updated!, unreadCount: Number(countRow?.count ?? 0) };
  });

  await publishRoomEvent(redis, {
    room: `u:${userId}`,
    event: {
      type: "read:updated",
      payload: { channelMember: toWireChannelMember(result.channelMember, result.unreadCount) },
    },
  });
  return { lastReadSeq: result.channelMember.lastReadSeq, unreadCount: result.unreadCount };
}

async function withReactionSummaries<T extends typeof schema.messages.$inferSelect>(
  tx: any,
  rows: T[],
  userId: number,
): Promise<Message[]> {
  if (rows.length === 0) return [];
  const ids = rows.map((row) => row.id);
  const reactionRows = await tx
    .select({
      messageId: schema.reactions.messageId,
      emoji: schema.reactions.emoji,
      userId: schema.reactions.userId,
    })
    .from(schema.reactions)
    .where(inArray(schema.reactions.messageId, ids));

  const referencedAttachmentIds = Array.from(new Set(rows.flatMap((row) => attachmentIdsFromBlocks(row.blocks))));
  const availableAttachmentIds = new Set<number>();
  if (referencedAttachmentIds.length > 0) {
    const attachmentRows = await tx.select({ id: schema.attachments.id }).from(schema.attachments).where(and(
      inArray(schema.attachments.id, referencedAttachmentIds),
      eq(schema.attachments.status, "ready"),
      isNull(schema.attachments.deletedAt),
    ));
    for (const attachment of attachmentRows) availableAttachmentIds.add(attachment.id);
  }

  const grouped = new Map<number, Map<string, { count: number; reactedByMe: boolean }>>();
  for (const reaction of reactionRows) {
    const byEmoji = grouped.get(reaction.messageId) ?? new Map<string, { count: number; reactedByMe: boolean }>();
    const current = byEmoji.get(reaction.emoji) ?? { count: 0, reactedByMe: false };
    current.count += 1;
    if (reaction.userId === userId) current.reactedByMe = true;
    byEmoji.set(reaction.emoji, current);
    grouped.set(reaction.messageId, byEmoji);
  }

  return rows.map((row) => toWireMessage(
    { ...row, blocks: removeUnavailableAttachments(row.blocks, availableAttachmentIds) },
    Array.from(grouped.get(row.id)?.entries() ?? []).map(([emoji, summary]) => ({ emoji, ...summary })),
  ));
}

/** A deleted message stays in place (so seq/ordering and reply counts don't
 * shift) but its content never leaves the server once `deletedAt` is set —
 * clients render a tombstone from `deletedAt` alone. Applied at every read
 * path; `message:deleted` broadcasts already carry no text (see
 * deleteMessage's payload). */
function scrubDeleted<T extends typeof schema.messages.$inferSelect>(row: T): T {
  if (!row.deletedAt) return row;
  return { ...row, text: "", blocks: { v: 1, doc: { type: "doc", content: [] } } };
}

// The DB columns behind `type`/`blocks` (messages) and `role`
// (channel_members) are plain text, not Postgres enums, so Drizzle's
// inferred row types are wider than the wire contract's literal unions.
// The cast documents that narrowing rather than hiding a real mismatch.
function toWireMessage(row: typeof schema.messages.$inferSelect, reactions: MessageReactionSummary[] = []): Message {
  return {
    id: row.id,
    workspaceId: row.workspaceId,
    channelId: row.channelId,
    seq: row.seq,
    clientMsgId: row.clientMsgId,
    authorId: row.authorId,
    type: row.type,
    text: row.text,
    blocks: row.blocks,
    revision: row.revision,
    parentId: row.parentId,
    isBroadcast: row.isBroadcast,
    threadReplyCount: row.threadReplyCount,
    threadLastReplyAt: row.threadLastReplyAt?.toISOString() ?? null,
    editedAt: row.editedAt?.toISOString() ?? null,
    deletedAt: row.deletedAt?.toISOString() ?? null,
    createdAt: row.createdAt.toISOString(),
    reactions,
  } as Message;
}

function toWireChannelMember(row: typeof schema.channelMembers.$inferSelect, unreadCount = 0): ChannelMember {
  return {
    channelId: row.channelId,
    userId: row.userId,
    role: row.role as ChannelMember["role"],
    joinedAt: row.joinedAt.toISOString(),
    lastReadSeq: row.lastReadSeq,
    lastReadAt: row.lastReadAt?.toISOString() ?? null,
    mentionCount: row.mentionCount,
    unreadCount,
    isMuted: row.isMuted,
    isStarred: row.isStarred,
    isClosed: row.isClosed,
  };
}