import {
  canPerformChatAction,
  chatCursorSchema,
  chatSequenceSchema,
} from "@tasks/shared";
import { and, asc, desc, eq, gt, inArray, isNull, lt, sql } from "drizzle-orm";
import { z } from "zod";

import { db as defaultDb } from "../client";
import {
  chatChannelMembers,
  chatChannels,
  chatMessageMentions,
  chatMessageMutations,
  chatMessages,
  chatOutbox,
  chatReadCursors,
  workspaceMembers,
} from "../schema";

export { chatCursorSchema, chatSequenceSchema } from "@tasks/shared";
export type ChatSequence = z.infer<typeof chatSequenceSchema>;
export type ChatCursor = z.infer<typeof chatCursorSchema>;

export const chatPageSizeSchema = z.number().int().min(1).max(100);
export const chatCatchUpSizeSchema = z.number().int().min(1).max(500);
const uuid = z.string().uuid();
const scopeSchema = z
  .object({ workspaceId: uuid, channelId: uuid, actorUserId: uuid })
  .strict();
const historySchema = scopeSchema
  .extend({ before: chatSequenceSchema.optional(), limit: chatPageSizeSchema })
  .strict();
const threadSchema = historySchema.extend({ rootMessageId: uuid }).strict();
const catchUpSchema = scopeSchema
  .extend({
    after: chatCursorSchema,
    through: chatCursorSchema.optional(),
    limit: chatCatchUpSizeSchema,
  })
  .strict()
  .refine(
    (input) =>
      input.through === undefined ||
      BigInt(input.after) <= BigInt(input.through),
    { message: "After exceeds through", path: ["through"] },
  );
const mutationsSchema = scopeSchema
  .extend({
    after: chatCursorSchema,
    through: chatCursorSchema.optional(),
    limit: chatCatchUpSizeSchema,
  })
  .strict()
  .refine(
    (input) =>
      input.through === undefined ||
      BigInt(input.after) <= BigInt(input.through),
    { path: ["through"] },
  );
const markReadSchema = scopeSchema
  .extend({ sequence: chatSequenceSchema })
  .strict();
const summarySchema = z
  .object({ workspaceId: uuid, actorUserId: uuid })
  .strict();
type Scope = z.infer<typeof scopeSchema>;
type Tx = Parameters<Parameters<typeof defaultDb.transaction>[0]>[0];
export class ChatReadError extends Error {
  constructor() {
    super("Chat channel not found");
  }
}

// The same workspace+channel+actor policy guards every read surface, including archived channels.
async function authorize(tx: Tx, scope: Scope) {
  const [actor] = await tx
    .select({ role: workspaceMembers.role })
    .from(workspaceMembers)
    .where(
      and(
        eq(workspaceMembers.workspaceId, scope.workspaceId),
        eq(workspaceMembers.userId, scope.actorUserId),
      ),
    )
    .limit(1);
  const [channel] = await tx
    .select()
    .from(chatChannels)
    .where(
      and(
        eq(chatChannels.workspaceId, scope.workspaceId),
        eq(chatChannels.id, scope.channelId),
      ),
    )
    .limit(1);
  if (!actor || !channel) throw new ChatReadError();
  const [member] = await tx
    .select({ userId: chatChannelMembers.userId })
    .from(chatChannelMembers)
    .where(
      and(
        eq(chatChannelMembers.workspaceId, scope.workspaceId),
        eq(chatChannelMembers.channelId, scope.channelId),
        eq(chatChannelMembers.userId, scope.actorUserId),
      ),
    )
    .limit(1);
  if (
    !canPerformChatAction({
      action: "read",
      actor: {
        userId: scope.actorUserId,
        workspaceId: scope.workspaceId,
        activeWorkspaceMember: true,
        role: actor.role,
      },
      context: { type: "workspace", workspaceId: scope.workspaceId },
      channel: {
        id: channel.id,
        workspaceId: channel.workspaceId,
        kind: channel.kind,
        archived: channel.archivedAt !== null,
        activeMember: !!member,
        ownerUserId: channel.ownerUserId,
      },
    })
  )
    throw new ChatReadError();
  return channel;
}

type Message = typeof chatMessages.$inferSelect;
function serialize(message: Message) {
  // Explicit projection prevents internal send-intent fingerprints (and future storage keys)
  // from leaking through history and catch-up DTOs.
  return {
    id: message.id,
    workspaceId: message.workspaceId,
    channelId: message.channelId,
    sequence: message.sequence.toString(),
    parentMessageId: message.parentMessageId,
    authorType: message.authorType,
    authorUserId: message.authorUserId,
    authorAgentId: message.authorAgentId,
    body: message.status === "deleted" ? "" : message.body,
    status: message.status,
    revision: message.revision,
    createdAt: message.createdAt,
    editedAt: message.editedAt,
    deletedAt: message.deletedAt,
  };
}
async function replyMetadata(tx: Tx, scope: Scope, roots: Message[]) {
  if (!roots.length)
    return new Map<
      string,
      { replyCount: number; latestReplySequence: string | null }
    >();
  const rows = await tx
    .select({
      parentMessageId: chatMessages.parentMessageId,
      count: sql<number>`count(*)::int`,
      latest: sql<bigint>`max(${chatMessages.sequence})`,
    })
    .from(chatMessages)
    .where(
      and(
        eq(chatMessages.workspaceId, scope.workspaceId),
        eq(chatMessages.channelId, scope.channelId),
        inArray(
          chatMessages.parentMessageId,
          roots.map((root) => root.id),
        ),
      ),
    )
    .groupBy(chatMessages.parentMessageId);
  return new Map(
    rows.map((row) => [
      row.parentMessageId!,
      { replyCount: row.count, latestReplySequence: row.latest.toString() },
    ]),
  );
}

// Backward root-page snapshot: limit+1 detects older roots, while reply metadata uses the same snapshot.
export async function chatHistory(
  db: typeof defaultDb,
  raw: z.input<typeof historySchema>,
) {
  const input = historySchema.parse(raw);
  return retryTransaction(
    db,
    async (tx) => {
      await tx.execute(sql`set transaction read only`);
      await authorize(tx, input);
      const rows = await tx
        .select()
        .from(chatMessages)
        .where(
          and(
            eq(chatMessages.workspaceId, input.workspaceId),
            eq(chatMessages.channelId, input.channelId),
            isNull(chatMessages.parentMessageId),
            input.before
              ? lt(chatMessages.sequence, BigInt(input.before))
              : undefined,
          ),
        )
        .orderBy(desc(chatMessages.sequence))
        .limit(input.limit + 1);
      const hasMore = rows.length > input.limit;
      const page = rows.slice(0, input.limit);
      const metadata = await replyMetadata(tx, input, page);
      return {
        messages: page.map((message) => ({
          ...serialize(message),
          replyCount: metadata.get(message.id)?.replyCount ?? 0,
          latestReplySequence:
            metadata.get(message.id)?.latestReplySequence ?? null,
        })),
        next: hasMore ? page.at(-1)!.sequence.toString() : null,
        hasMore,
      };
    },
    { isolationLevel: "repeatable read" },
  );
}

// Thread pages include the root and only its direct replies; a reply is never a root.
export async function chatThreadHistory(
  db: typeof defaultDb,
  raw: z.input<typeof threadSchema>,
) {
  const input = threadSchema.parse(raw);
  return retryTransaction(
    db,
    async (tx) => {
      await tx.execute(sql`set transaction read only`);
      await authorize(tx, input);
      const [root] = await tx
        .select()
        .from(chatMessages)
        .where(
          and(
            eq(chatMessages.workspaceId, input.workspaceId),
            eq(chatMessages.channelId, input.channelId),
            eq(chatMessages.id, input.rootMessageId),
            isNull(chatMessages.parentMessageId),
          ),
        )
        .limit(1);
      if (!root) throw new ChatReadError();
      const rows = await tx
        .select()
        .from(chatMessages)
        .where(
          and(
            eq(chatMessages.workspaceId, input.workspaceId),
            eq(chatMessages.channelId, input.channelId),
            eq(chatMessages.parentMessageId, root.id),
            input.before
              ? lt(chatMessages.sequence, BigInt(input.before))
              : undefined,
          ),
        )
        .orderBy(desc(chatMessages.sequence))
        .limit(input.limit + 1);
      const hasMore = rows.length > input.limit;
      const page = rows.slice(0, input.limit);
      const metadata = await replyMetadata(tx, input, [root]);
      return {
        root: {
          ...serialize(root),
          replyCount: metadata.get(root.id)?.replyCount ?? 0,
          latestReplySequence:
            metadata.get(root.id)?.latestReplySequence ?? null,
        },
        replies: page.map(serialize),
        next: hasMore ? page.at(-1)!.sequence.toString() : null,
        hasMore,
      };
    },
    { isolationLevel: "repeatable read" },
  );
}

// Bounded reconnect scan includes replies and tombstones in channel sequence order.
export async function chatCatchUp(
  db: typeof defaultDb,
  raw: z.input<typeof catchUpSchema>,
) {
  const input = catchUpSchema.parse(raw);
  return retryTransaction(
    db,
    async (tx) => {
      await tx.execute(sql`set transaction read only`);
      const channel = await authorize(tx, input);
      const highWater = channel.nextSequence - 1n;
      if (
        BigInt(input.after) > highWater ||
        (input.through !== undefined && BigInt(input.through) > highWater)
      )
        throw new ChatReadError();
      const through = input.through ?? highWater.toString();
      const rows = await tx
        .select()
        .from(chatMessages)
        .where(
          and(
            eq(chatMessages.workspaceId, input.workspaceId),
            eq(chatMessages.channelId, input.channelId),
            gt(chatMessages.sequence, BigInt(input.after)),
            sql`${chatMessages.sequence} <= ${BigInt(through)}`,
          ),
        )
        .orderBy(asc(chatMessages.sequence))
        .limit(input.limit + 1);
      const hasMore = rows.length > input.limit;
      const page = rows.slice(0, input.limit);
      const roots = page.filter((message) => message.parentMessageId === null);
      const metadata = await replyMetadata(tx, input, roots);
      return {
        messages: page.map((message) => ({
          ...serialize(message),
          replyCount:
            message.parentMessageId === null
              ? (metadata.get(message.id)?.replyCount ?? 0)
              : 0,
          latestReplySequence:
            message.parentMessageId === null
              ? (metadata.get(message.id)?.latestReplySequence ?? null)
              : null,
        })),
        next: page.at(-1)?.sequence.toString() ?? input.after,
        through,
        hasMore,
      };
    },
    { isolationLevel: "repeatable read" },
  );
}

// Durable metadata-only mutation scan: fixed high-water survives pagination and reconnects.
export async function chatMutationCatchUp(
  db: typeof defaultDb,
  raw: z.input<typeof mutationsSchema>,
) {
  const input = mutationsSchema.parse(raw);
  return retryTransaction(
    db,
    async (tx) => {
      await tx.execute(sql`set transaction read only`);
      const channel = await authorize(tx, input);
      const highWater = channel.nextMutationSequence - 1n;
      if (
        BigInt(input.after) > highWater ||
        (input.through !== undefined && BigInt(input.through) > highWater)
      )
        throw new ChatReadError();
      const through = input.through ?? highWater.toString();
      const rows = await tx
        .select({
          messageId: chatMessageMutations.messageId,
          mutationSequence: chatMessageMutations.mutationSequence,
          revision: chatMessageMutations.revision,
          kind: chatMessageMutations.kind,
        })
        .from(chatMessageMutations)
        .where(
          and(
            eq(chatMessageMutations.workspaceId, input.workspaceId),
            eq(chatMessageMutations.channelId, input.channelId),
            gt(chatMessageMutations.mutationSequence, BigInt(input.after)),
            sql`${chatMessageMutations.mutationSequence} <= ${BigInt(through)}`,
          ),
        )
        .orderBy(asc(chatMessageMutations.mutationSequence))
        .limit(input.limit + 1);
      const page = rows.slice(0, input.limit);
      const messageIds = [...new Set(page.map((row) => row.messageId))];
      const messagesRows = messageIds.length ? await tx
        .select()
        .from(chatMessages)
        .where(
          and(
            eq(chatMessages.workspaceId, input.workspaceId),
            eq(chatMessages.channelId, input.channelId),
            inArray(chatMessages.id, messageIds),
          ),
        ) : [];
      const messagesMap = new Map(messagesRows.map((m) => [m.id, serialize(m)]));
      if (messagesMap.size !== messageIds.length) {
        throw new ChatReadError();
      }
      return {
        mutations: page.map((row) => ({
          messageId: row.messageId,
          mutationSequence: row.mutationSequence.toString(),
          revision: row.revision,
          kind: row.kind,
        })),
        messages: messageIds.map((id) => messagesMap.get(id)!),
        next: page.at(-1)?.mutationSequence.toString() ?? input.after,
        through,
        hasMore: rows.length > input.limit,
      };
    },
    { isolationLevel: "repeatable read" },
  );
}

// Mark-read validates that the claimed viewed sequence actually exists, then advances and emits
// exactly one invalidation in the same transaction. A no-op never changes the cursor or emits.
export async function chatMarkChannelRead(
  db: typeof defaultDb,
  raw: z.input<typeof markReadSchema>,
) {
  const input = markReadSchema.parse(raw);
  return retryTransaction(
    db,
    async (tx) => {
      await authorize(tx, input);
      const [viewed] = await tx
        .select({ id: chatMessages.id })
        .from(chatMessages)
        .where(
          and(
            eq(chatMessages.workspaceId, input.workspaceId),
            eq(chatMessages.channelId, input.channelId),
            eq(chatMessages.sequence, BigInt(input.sequence)),
          ),
        )
        .limit(1);
      if (!viewed) throw new ChatReadError();
      const [current] = await tx
        .select({ sequence: chatReadCursors.sequence })
        .from(chatReadCursors)
        .where(
          and(
            eq(chatReadCursors.workspaceId, input.workspaceId),
            eq(chatReadCursors.channelId, input.channelId),
            eq(chatReadCursors.userId, input.actorUserId),
          ),
        )
        .limit(1);
      if (current && current.sequence >= BigInt(input.sequence))
        return { sequence: current.sequence.toString(), changed: false };
      const [updated] = await tx
        .insert(chatReadCursors)
        .values({
          workspaceId: input.workspaceId,
          channelId: input.channelId,
          userId: input.actorUserId,
          sequence: BigInt(input.sequence),
        })
        .onConflictDoUpdate({
          target: [
            chatReadCursors.workspaceId,
            chatReadCursors.channelId,
            chatReadCursors.userId,
          ],
          set: {
            sequence: sql`greatest(${chatReadCursors.sequence}, excluded.sequence)`,
            updatedAt: sql`now()`,
          },
          setWhere: lt(chatReadCursors.sequence, BigInt(input.sequence)),
        })
        .returning({ sequence: chatReadCursors.sequence });
      if (!updated)
        return {
          sequence: current?.sequence.toString() ?? "0",
          changed: false,
        };
      await tx.insert(chatOutbox).values({
        workspaceId: input.workspaceId,
        channelId: input.channelId,
        eventType: "chat.read_state.changed.v1",
        eventVersion: 1,
        dedupeKey: `read:${input.channelId}:${input.actorUserId}:${updated.sequence}`,
        payload: {
          eventType: "chat.read_state.changed.v1",
          eventVersion: 1,
          workspaceId: input.workspaceId,
          channelId: input.channelId,
          userId: input.actorUserId,
          sequence: updated.sequence.toString(),
        },
      });
      return { sequence: updated.sequence.toString(), changed: true };
    },
    { isolationLevel: "repeatable read" },
  );
}

// Retry the transaction as a unit so a failed attempt cannot commit a cursor without its outbox event.
async function retryTransaction<T>(
  db: typeof defaultDb,
  callback: (tx: Tx) => Promise<T>,
  config: Parameters<typeof defaultDb.transaction>[1],
): Promise<T> {
  for (let attempt = 0; attempt < 3; attempt++) {
    try {
      return await db.transaction((tx) => callback(tx), config);
    } catch (error) {
      if (attempt === 2 || !isRetryableTransactionError(error)) throw error;
    }
  }
  throw new Error("Unreachable transaction retry state");
}

function isRetryableTransactionError(error: unknown): boolean {
  if (typeof error !== "object" || error === null) return false;
  const candidate = error as { code?: unknown; cause?: unknown };
  return (
    candidate.code === "40001" ||
    candidate.code === "40P01" ||
    (candidate.cause !== undefined &&
      isRetryableTransactionError(candidate.cause))
  );
}

// Unread aggregation considers every authorized channel, including archived ones, but never
// grants private/direct visibility based on the actor's workspace role alone.
export async function chatUnreadSummary(
  db: typeof defaultDb,
  raw: z.input<typeof summarySchema>,
) {
  const input = summarySchema.parse(raw);
  return db.transaction(
    async (tx) => {
      await tx.execute(sql`set transaction read only`);
      const [actor] = await tx
        .select({ userId: workspaceMembers.userId })
        .from(workspaceMembers)
        .where(
          and(
            eq(workspaceMembers.workspaceId, input.workspaceId),
            eq(workspaceMembers.userId, input.actorUserId),
          ),
        )
        .limit(1);
      if (!actor) throw new ChatReadError();
      const channels = await tx
        .select({ channel: chatChannels })
        .from(chatChannels)
        .leftJoin(
          chatChannelMembers,
          and(
            eq(chatChannelMembers.workspaceId, chatChannels.workspaceId),
            eq(chatChannelMembers.channelId, chatChannels.id),
            eq(chatChannelMembers.userId, input.actorUserId),
          ),
        )
        .where(
          and(
            eq(chatChannels.workspaceId, input.workspaceId),
            sql`(${chatChannels.kind} = 'public' or ${chatChannelMembers.userId} is not null)`,
          ),
        );
      // A single set-based count query keeps the summary bounded by channels, not messages.
      const counts = await tx
        .select({
          channelId: chatChannels.id,
          readSequence: sql<bigint>`coalesce(${chatReadCursors.sequence}, 0)`,
          unreadCount: sql<number>`count(${chatMessages.id})::int`,
          mentionCount: sql<number>`count(${chatMessageMentions.messageId})::int`,
        })
        .from(chatChannels)
        .leftJoin(
          chatReadCursors,
          and(
            eq(chatReadCursors.workspaceId, chatChannels.workspaceId),
            eq(chatReadCursors.channelId, chatChannels.id),
            eq(chatReadCursors.userId, input.actorUserId),
          ),
        )
        .leftJoin(
          chatMessages,
          and(
            eq(chatMessages.workspaceId, chatChannels.workspaceId),
            eq(chatMessages.channelId, chatChannels.id),
            sql`${chatMessages.sequence} > coalesce(${chatReadCursors.sequence}, 0)`,
          ),
        )
        .leftJoin(
          chatMessageMentions,
          and(
            eq(chatMessageMentions.workspaceId, chatMessages.workspaceId),
            eq(chatMessageMentions.messageId, chatMessages.id),
            eq(chatMessageMentions.userId, input.actorUserId),
          ),
        )
        .where(
          and(
            eq(chatChannels.workspaceId, input.workspaceId),
            inArray(
              chatChannels.id,
              channels.map(({ channel }) => channel.id),
            ),
          ),
        )
        .groupBy(chatChannels.id, chatReadCursors.sequence);
      const byId = new Map(counts.map((count) => [count.channelId, count]));
      return {
        channels: channels.map(({ channel }) => ({
          channelId: channel.id,
          archived: channel.archivedAt !== null,
          readSequence: byId.get(channel.id)?.readSequence.toString() ?? "0",
          unreadCount: byId.get(channel.id)?.unreadCount ?? 0,
          mentionCount: byId.get(channel.id)?.mentionCount ?? 0,
        })),
      };
    },
    { isolationLevel: "repeatable read" },
  );
}