import {
  findById,
  patchById,
  removeById,
} from '@crumbs/core/collections'
import {
  DmConversationSchema,
  DmMessageSchema,
  type DmConversation,
  type DmMessage,
} from '@crumbs/core/schemas'
import {create} from 'zustand'

import {createSlotStore, slotHooks} from '@/lib/async-slot'
import {isApiStatus, unwrap} from '@/lib/errors'
import {createKeyedPagedStore} from '@/lib/keyed-paged-resource'
import {pageParams} from '@/lib/paged-resource'
import {rpc} from '@/lib/rpc'
import {fromSchema} from '@/lib/snapshot-cache'
import {keyedPagedSnapshot, listSnapshot} from '@/lib/snapshots'
import {registerReset} from '@/lib/store-reset'
import {uuid} from '@/lib/uuid'
import {requireUser} from '@/stores/auth'
import {defineMutation} from '@/stores/outbox'

export type {DmConversation, DmMessage}

// the inbox: unpaginated, server-ordered by latest activity
const useConversationsStore = createSlotStore<DmConversation[]>(
  [],
  async () => {
    const {conversations} = await unwrap(rpc.dms.$get())
    return conversations
  },
  // v2: peerIsFriend became a required field
  listSnapshot(
    'dms-conversations',
    2,
    fromSchema(DmConversationSchema),
  ),
)

// newest server-returned createdAt per conversation: pollNew's `since`
// cursor. advanced only by server reads (the initial page, poll results),
// never by the slice head — an offline send reconciles to an ack-time
// createdAt later than peer messages sent while offline, and deriving
// `since` from it would skip those peer messages forever
const serverCursor = new Map<string, string>()

const advanceServerCursor = (
  conversationId: string,
  createdAt: string | undefined,
): void => {
  if (!createdAt) return
  const current = serverCursor.get(conversationId)
  if (!current || createdAt > current)
    serverCursor.set(conversationId, createdAt)
}

// per-conversation history, newest first (rendered by an inverted list)
const messagesPages = createKeyedPagedStore<DmMessage>(
  (conversationId) => async (cursor) => {
    const {messages, nextCursor} = await unwrap(
      rpc.dms[':id'].messages.$get({
        param: {id: conversationId},
        query: pageParams(cursor),
      }),
    )
    if (!cursor)
      advanceServerCursor(conversationId, messages[0]?.createdAt)
    return {items: messages, nextCursor}
  },
  keyedPagedSnapshot('dms-messages', 1, fromSchema(DmMessageSchema)),
)

const messageId = (m: DmMessage) => m.id

const patchConversation = (
  conversationId: string,
  fn: (c: DmConversation) => DmConversation,
): void =>
  useConversationsStore.setState((s) => ({
    data: patchById(s.data, conversationId, fn),
  }))

// drop a message everywhere it may show: the thread slice and, when it was
// the inbox row's preview, the row itself (recomputed from the slice head)
const dropMessage = (conversationId: string, id: string): void => {
  messagesPages.removeItem((m) => m.id === id)
  const [newest = null] = messagesPages.getItems(conversationId)
  patchConversation(conversationId, (c) =>
    c.lastMessage?.id === id
      ? {
          ...c,
          lastMessage: newest,
          lastMessageAt: newest?.createdAt ?? null,
        }
      : c,
  )
}

// the friendship ended locally (the viewer unfriended the peer): close the
// channel at once so an open thread swaps its composer for the read-only
// bar without waiting on an inbox refresh
export const closeConversationWithPeer = (peerId: string): void =>
  useConversationsStore.setState((s) => ({
    data: s.data.map((c) =>
      c.peer.id === peerId ? {...c, peerIsFriend: false} : c,
    ),
  }))

// a new message moves its conversation to the top of the inbox
const bumpConversation = (message: DmMessage): void =>
  useConversationsStore.setState((s) => {
    const existing = findById(s.data, message.conversationId)
    if (!existing) return s
    const updated = {
      ...existing,
      lastMessage: message,
      lastMessageAt: message.createdAt,
    }
    return {data: [updated, ...removeById(s.data, updated.id)]}
  })

// send is scoped by conversation id so a burst of offline messages replays in
// order (create-before-create within a conversation); different conversations
// replay in parallel
const DM_SEND = 'dm.send'

type DmsState = {
  // the conversation currently on screen; the notification handler reads
  // it to suppress banners for messages the user is already looking at
  activeConversationId: string | null
  setActiveConversation: (id: string | null) => void
  openConversation: (userId: string) => Promise<string>
  send: (conversationId: string, text: string) => Promise<void>
  pollNew: (conversationId: string) => Promise<void>
  markRead: (conversationId: string) => Promise<void>
}

export const useDmsStore = create<DmsState>((set, get) => ({
  activeConversationId: null,
  setActiveConversation: (id) => set({activeConversationId: id}),
  openConversation: async (userId) => {
    const {conversation} = await unwrap(
      rpc.dms.$post({json: {userId}}),
    )
    useConversationsStore.setState((s) =>
      s.data.some((c) => c.id === conversation.id)
        ? s
        : {data: [conversation, ...s.data]},
    )
    return conversation.id
  },
  send: async (conversationId, text) => {
    const user = requireUser()
    // mint the id so the optimistic bubble and the queued send share it: a
    // replayed send upserts on this pk, so a lost ack never doubles it. the
    // bubble shows at once (offline too) and the outbox replays the real send
    const id = uuid()
    const message: DmMessage = {
      id,
      conversationId,
      senderId: user.id,
      text,
      createdAt: new Date().toISOString(),
    }
    messagesPages.prependItems(conversationId, [message], messageId)
    bumpConversation(message)
    queueDmSend({
      scope: conversationId,
      id,
      label: text,
      payload: {id, conversationId, text},
    })
  },
  pollNew: async (conversationId) => {
    // the initial load owns the first fetch (and seeds the cursor);
    // polling is incremental only
    const since = serverCursor.get(conversationId)
    if (!since) return
    const {messages} = await unwrap(
      rpc.dms[':id'].messages.$get({
        param: {id: conversationId},
        query: {since},
      }),
    )
    if (!messages.length) return
    // since returns ascending; the slice holds newest first
    messagesPages.prependItems(
      conversationId,
      [...messages].reverse(),
      messageId,
    )
    const latest = messages[messages.length - 1]
    advanceServerCursor(conversationId, latest?.createdAt)
    if (latest) bumpConversation(latest)
    void get()
      .markRead(conversationId)
      .catch(() => {})
  },
  markRead: async (conversationId) => {
    patchConversation(conversationId, (c) => ({
      ...c,
      unreadCount: 0,
    }))
    await unwrap(
      rpc.dms[':id'].read.$put({param: {id: conversationId}}),
    )
  },
}))

registerReset(() => {
  useDmsStore.setState({activeConversationId: null})
  serverCursor.clear()
})

// reconcile the optimistic bubble to the server truth on ack: same id and
// text, but the canonical createdAt, so the thread orders by server time.
// deliberately does not advance the poll cursor — the ack proves nothing
// about peer messages sent while this one was queued
const applyServerMessage = (message: DmMessage): void => {
  // an in-flight refresh may have dropped the optimistic bubble before the
  // ack: re-insert it (dedupes by id, so the normal path no-ops)
  messagesPages.prependItems(
    message.conversationId,
    [message],
    messageId,
  )
  messagesPages.mapItems((m) => (m.id === message.id ? message : m))
  patchConversation(message.conversationId, (c) =>
    c.lastMessage?.id === message.id
      ? {...c, lastMessage: message, lastMessageAt: message.createdAt}
      : c,
  )
}

// the outbox replays each queued send with its client id (idempotent under a
// lost ack) and reconciles the optimistic bubble to the server copy
const queueDmSend = defineMutation<{
  id: string
  conversationId: string
  text: string
}>(DM_SEND, {
  run: async ({id, conversationId, text}) => {
    try {
      const {message} = await unwrap(
        rpc.dms[':id'].messages.$post({
          param: {id: conversationId},
          json: {id, text},
        }),
      )
      applyServerMessage(message)
    } catch (err) {
      // the channel closed (unfriended) between queueing and sync — a 403 the
      // send route only throws for not_friends. the message can never land, so
      // drop it and its optimistic bubble instead of dead-lettering, which
      // would wedge the rest of the thread behind it. flipping peerIsFriend
      // surfaces the read-only bar in place of the composer at once
      if (isApiStatus(err, 403)) {
        dropMessage(conversationId, id)
        patchConversation(conversationId, (c) => ({
          ...c,
          peerIsFriend: false,
        }))
        return
      }
      throw err
    }
  },
  // a send that can never land (dead-lettered then discarded): drop the bubble
  rollback: ({id, conversationId}) => dropMessage(conversationId, id),
})

export const getActiveConversationId = (): string | null =>
  useDmsStore.getState().activeConversationId

export const {
  useData: useConversations,
  useLoading: useConversationsLoading,
  useError: useConversationsError,
  useRefresh: useRefreshConversations,
} = slotHooks(useConversationsStore)
export const useConversation = (id: string) =>
  useConversationsStore((s) => findById(s.data, id))
export const useTotalUnread = () =>
  useConversationsStore((s) =>
    s.data.reduce((sum, c) => sum + c.unreadCount, 0),
  )

export const useMessages = messagesPages.useItems
export const useMessagesLoading = messagesPages.useLoading
export const useMessagesLoadingMore = messagesPages.useLoadingMore
export const useMessagesReachedEnd = messagesPages.useReachedEnd
export const useLoadMessages = messagesPages.useLoad
export const useLoadMoreMessages = messagesPages.useLoadMore

export const useOpenConversation = () =>
  useDmsStore((s) => s.openConversation)
export const useSendMessage = () => useDmsStore((s) => s.send)
export const usePollNewMessages = () => useDmsStore((s) => s.pollNew)
export const useMarkConversationRead = () =>
  useDmsStore((s) => s.markRead)
export const useSetActiveConversation = () =>
  useDmsStore((s) => s.setActiveConversation)

export const refreshConversations = () =>
  useConversationsStore.getState().refresh()
