import {
  MAX_PAGE_SIZE,
  type DmConversation,
  type DmMessage,
} from '@crumbs/core/schemas'
import {and, asc, desc, eq, inArray, or, sql} from 'drizzle-orm'
import {HTTPException} from 'hono/http-exception'

import {db, schema} from '../db/client'
import {
  keysetCursorWhere,
  keysetOrder,
  ms,
  toPage,
} from '../lib/cursor'
import {notBlockedByViewer} from './blocks'
import {listFriendIds} from './friends'
import {loadUserSummaries} from './users'

type ConversationRow = typeof schema.conversations.$inferSelect
type MessageRow = typeof schema.messages.$inferSelect

type ConversationSide = 'a' | 'b'

export const canonicalPair = (
  a: string,
  b: string,
): [string, string] => (a < b ? [a, b] : [b, a])

export const toDmMessage = (row: MessageRow): DmMessage => ({
  id: row.id,
  conversationId: row.conversationId,
  senderId: row.senderId,
  text: row.text,
  createdAt: row.createdAt.toISOString(),
})

export const peerIdFor = (
  conversation: ConversationRow,
  viewerId: string,
): string =>
  conversation.userAId === viewerId
    ? conversation.userBId
    : conversation.userAId

// participant lookup; 404 for both missing and foreign conversations so
// existence never leaks (mirrors requireTrailMember)
export const requireConversationMember = async (
  id: string,
  userId: string,
): Promise<{
  conversation: ConversationRow
  side: ConversationSide
}> => {
  const [conversation] = await db
    .select()
    .from(schema.conversations)
    .where(eq(schema.conversations.id, id))
    .limit(1)
  const side: ConversationSide | null = !conversation
    ? null
    : conversation.userAId === userId
      ? 'a'
      : conversation.userBId === userId
        ? 'b'
        : null
  if (!conversation || !side)
    throw new HTTPException(404, {message: 'not_found'})
  return {conversation, side}
}

export const getOrCreateConversation = async (
  viewerId: string,
  otherId: string,
): Promise<ConversationRow> => {
  const [userAId, userBId] = canonicalPair(viewerId, otherId)
  const [created] = await db
    .insert(schema.conversations)
    .values({userAId, userBId})
    .onConflictDoNothing()
    .returning()
  if (created) return created
  const [existing] = await db
    .select()
    .from(schema.conversations)
    .where(
      and(
        eq(schema.conversations.userAId, userAId),
        eq(schema.conversations.userBId, userBId),
      ),
    )
    .limit(1)
  if (!existing) throw new Error('failed to create conversation')
  return existing
}

// correlated column refs are spelled out: drizzle renders schema columns
// unqualified here, which the messages alias would shadow
const unreadCountSql = (viewerId: string) => sql<number>`(
  select count(*) from messages m
  where m.conversation_id = conversations.id
    and m.sender_id <> ${viewerId}
    and m.created_at > coalesce(
      case when conversations.user_a_id = ${viewerId}
        then conversations.user_a_last_read_at
        else conversations.user_b_last_read_at
      end,
      'epoch'::timestamptz
    )
)::int`

// the viewer's inbox, newest activity first. unpaginated: the list is
// bounded by friend count, capped defensively
export const listConversations = async (
  viewerId: string,
  conversationId?: string,
): Promise<DmConversation[]> => {
  const rows = await db
    .select({
      conversation: schema.conversations,
      unreadCount: unreadCountSql(viewerId).as('unread_count'),
    })
    .from(schema.conversations)
    .where(
      and(
        or(
          eq(schema.conversations.userAId, viewerId),
          eq(schema.conversations.userBId, viewerId),
        ),
        // hide conversations with a blocked peer (the viewer's own side of
        // the pair is trivially unblocked, so this reduces to the peer)
        notBlockedByViewer(schema.conversations.userAId, viewerId),
        notBlockedByViewer(schema.conversations.userBId, viewerId),
        conversationId
          ? eq(schema.conversations.id, conversationId)
          : undefined,
      ),
    )
    .orderBy(
      sql`${schema.conversations.lastMessageAt} desc nulls last`,
      desc(schema.conversations.createdAt),
    )
    .limit(MAX_PAGE_SIZE)
  if (!rows.length) return []

  const conversationIds = rows.map((r) => r.conversation.id)
  const peerIds = rows.map((r) => peerIdFor(r.conversation, viewerId))
  const [lastMessages, peers, friendIds] = await Promise.all([
    db
      .selectDistinctOn([schema.messages.conversationId])
      .from(schema.messages)
      .where(inArray(schema.messages.conversationId, conversationIds))
      .orderBy(
        asc(schema.messages.conversationId),
        desc(schema.messages.createdAt),
        desc(schema.messages.id),
      ),
    loadUserSummaries(peerIds),
    listFriendIds(viewerId),
  ])
  const lastByConversation = new Map(
    lastMessages.map((m) => [m.conversationId, m]),
  )
  const friendIdSet = new Set(friendIds)

  return rows.flatMap(({conversation, unreadCount}) => {
    const peer = peers.get(peerIdFor(conversation, viewerId))
    if (!peer) return []
    const last = lastByConversation.get(conversation.id)
    return [
      {
        id: conversation.id,
        peer,
        lastMessage: last ? toDmMessage(last) : null,
        lastMessageAt:
          conversation.lastMessageAt?.toISOString() ?? null,
        unreadCount,
        peerIsFriend: friendIdSet.has(peer.id),
        createdAt: conversation.createdAt.toISOString(),
      },
    ]
  })
}

export const getConversationView = async (
  viewerId: string,
  conversationId: string,
): Promise<DmConversation> => {
  const [view] = await listConversations(viewerId, conversationId)
  if (!view) throw new HTTPException(404, {message: 'not_found'})
  return view
}

// keyset-paginated history, newest first
export const listMessages = async (
  conversationId: string,
  {limit, cursor}: {limit: number; cursor?: string},
) => {
  const rows = await db
    .select()
    .from(schema.messages)
    .where(
      and(
        eq(schema.messages.conversationId, conversationId),
        keysetCursorWhere(
          schema.messages.createdAt,
          schema.messages.id,
          cursor,
        ),
      ),
    )
    .orderBy(
      ...keysetOrder(schema.messages.createdAt, schema.messages.id),
    )
    .limit(limit + 1)

  const {page, nextCursor} = toPage(rows, limit, (r) => ({
    createdAt: r.createdAt,
    id: r.id,
  }))
  return {messages: page.map(toDmMessage), nextCursor}
}

// incremental fetch for the open-conversation poll. compared at millisecond
// precision to match the ISO timestamps clients hold; messages sharing the
// boundary millisecond are caught by client-side id dedupe on a later poll
export const listMessagesSince = async (
  conversationId: string,
  since: Date,
): Promise<DmMessage[]> => {
  const rows = await db
    .select()
    .from(schema.messages)
    .where(
      and(
        eq(schema.messages.conversationId, conversationId),
        sql`${ms(schema.messages.createdAt)} > ${since}`,
      ),
    )
    .orderBy(asc(schema.messages.createdAt), asc(schema.messages.id))
    .limit(MAX_PAGE_SIZE)
  return rows.map(toDmMessage)
}

export const markRead = async (
  conversationId: string,
  side: ConversationSide,
  at: Date = new Date(),
): Promise<void> => {
  await db
    .update(schema.conversations)
    .set(side === 'a' ? {userALastReadAt: at} : {userBLastReadAt: at})
    .where(eq(schema.conversations.id, conversationId))
}
