Pawer/cloudfunctions/messageSend/index.js

const cloud = require('wx-server-sdk')

cloud.init({ env: cloud.DYNAMIC_CURRENT_ENV })
const db = cloud.database()
const _ = db.command
const SERVER_MESSAGE_LIMIT = 500
const CONVERSATIONS_COLLECTION = 'conversations'
const MESSAGES_COLLECTION = 'messages'
const CHAT_MESSAGE_TEMPLATE_ID = process.env.WQ_MESSAGE_TEMPLATE_ID || process.env.TARO_APP_MESSAGE_TEMPLATE_ID || ''

function isCollectionMissing(error) {
  const text = String(error?.errMsg || error?.message || error || '')
  return (
    text.includes('collection not exist') ||
    text.includes('collection not exists') ||
    text.includes('collection is not exists') ||
    text.includes('DATABASE_COLLECTION_NOT_EXIST') ||
    text.includes('-502005')
  )
}

function isCollectionAlreadyExists(error) {
  const text = String(error?.errMsg || error?.message || error || '')
  return text.includes('collection already exists') || text.includes('DATABASE_COLLECTION_ALREADY_EXIST')
}

async function ensureCollection(name) {
  if (typeof db.createCollection !== 'function') {
    throw new Error(`${name} collection does not exist`)
  }

  try {
    await db.createCollection(name)
  } catch (error) {
    if (!isCollectionAlreadyExists(error)) throw error
  }
}

async function getExistingConversation(openid, toOpenid) {
  try {
    return await db.collection(CONVERSATIONS_COLLECTION)
      .where({ type: 'single', memberOpenids: _.all([openid, toOpenid]) })
      .limit(1)
      .get()
  } catch (error) {
    if (!isCollectionMissing(error)) throw error
    await ensureCollection(CONVERSATIONS_COLLECTION)
    return { data: [] }
  }
}

async function getConversationById(conversationId) {
  try {
    return await db.collection(CONVERSATIONS_COLLECTION).doc(conversationId).get()
  } catch (error) {
    if (!isCollectionMissing(error)) throw error
    return { data: null }
  }
}

async function addConversation(data) {
  try {
    return await db.collection(CONVERSATIONS_COLLECTION).add({ data })
  } catch (error) {
    if (!isCollectionMissing(error)) throw error
    await ensureCollection(CONVERSATIONS_COLLECTION)
    return db.collection(CONVERSATIONS_COLLECTION).add({ data })
  }
}

async function addMessage(data) {
  try {
    return await db.collection(MESSAGES_COLLECTION).add({ data })
  } catch (error) {
    if (!isCollectionMissing(error)) throw error
    await ensureCollection(MESSAGES_COLLECTION)
    return db.collection(MESSAGES_COLLECTION).add({ data })
  }
}

async function trimConversationMessages(conversationId) {
  for (;;) {
    const oldMessages = await db.collection(MESSAGES_COLLECTION)
      .where({ conversationId })
      .orderBy('createdAt', 'desc')
      .skip(SERVER_MESSAGE_LIMIT)
      .limit(100)
      .get()
      .catch(() => ({ data: [] }))

    if (!oldMessages.data.length) return

    await Promise.all(
      oldMessages.data.map(item =>
        db.collection(MESSAGES_COLLECTION).doc(item._id).remove().catch(() => null)
      )
    )

    if (oldMessages.data.length < 100) return
  }
}

function pad(value) {
  return String(value).padStart(2, '0')
}

function formatDateTime(ts) {
  const d = new Date(ts)
  return `${d.getFullYear()}-${pad(d.getMonth() + 1)}-${pad(d.getDate())} ${pad(d.getHours())}:${pad(d.getMinutes())}`
}

function templateValue(value, max = 20) {
  const text = String(value || '').trim()
  return text.length > max ? `${text.slice(0, max - 1)}…` : text
}

async function sendChatNotifications({ recipients, conversationId, senderName, content, type, createdAt }) {
  if (!CHAT_MESSAGE_TEMPLATE_ID || !recipients.length || !cloud.openapi?.subscribeMessage?.send) return

  const preview = type === 'image' ? '[图片]' : content
  await Promise.all(
    recipients.map(touser =>
      cloud.openapi.subscribeMessage.send({
        touser,
        templateId: CHAT_MESSAGE_TEMPLATE_ID,
        page: `pages/chat/index?conversationId=${conversationId}`,
        data: {
          thing31: { value: templateValue(senderName || '汪友') },
          thing2: { value: templateValue(preview || '你收到一条新消息') },
          time3: { value: formatDateTime(createdAt) }
        }
      }).catch(error => {
        console.log('[messageSend] subscribe message skipped', error?.errMsg || error?.message || error)
      })
    )
  )
}

exports.main = async event => {
  const { OPENID } = cloud.getWXContext()
  const content = String(event.content || '').trim()
  if (!content) throw new Error('content is required')
  if (!OPENID) throw new Error('no openid in wx context')

  const now = Date.now()
  let convId = event.conversationId
  let conversation = null

  // Create a 1:1 conversation if not provided
  if (!convId) {
    let { toOpenid } = event
    // Allow opening by user id without exposing openids to the client.
    if (!toOpenid && event.toUserId) {
      const toUserById = await db.collection('users').doc(event.toUserId).get().catch(() => null)
      if (toUserById && toUserById.data) toOpenid = toUserById.data.openid
      if (!toOpenid) {
        const toUserByOpenid = await db.collection('users')
          .where({ openid: event.toUserId })
          .limit(1)
          .get()
          .catch(() => ({ data: [] }))
        if (toUserByOpenid.data[0]) toOpenid = toUserByOpenid.data[0].openid
      }
    }
    if (!toOpenid) throw new Error('target user not found')

    // Look for existing conversation between these two users
    const existing = await getExistingConversation(OPENID, toOpenid)

    if (existing.data.length) {
      conversation = existing.data[0]
      convId = conversation._id
    } else {
      const toUserRes = await db.collection('users').where({ openid: toOpenid }).limit(1).get()
      const toUser = toUserRes.data[0] || { nickname: '用户', avatarKey: 'gradient-avatar-1' }
      conversation = {
        type: 'single',
        memberOpenids: [OPENID, toOpenid],
        unreadOpenids: [],
        unreadCounts: {},
        title: toUser.nickname,
        avatarKey: toUser.avatarKey,
        lastMessage: { content: content.slice(0, 50), createdAt: now },
        pinned: false,
        muted: false,
        createdAt: now,
        updatedAt: now
      }
      const newConv = await addConversation(conversation)
      convId = newConv._id
    }
  }

  if (!conversation) {
    const convRes = await getConversationById(convId)
    conversation = convRes.data
  }
  if (!conversation) throw new Error('conversation not found')

  const members = conversation.memberOpenids || []
  if (!members.includes(OPENID)) throw new Error('permission denied')
  if ((conversation.type || 'single') !== 'single') throw new Error('unsupported conversation type')

  // Add message to messages collection
  const msgResult = await addMessage({
    conversationId: convId,
    fromOpenid: OPENID,
    content: content.slice(0, 1000),
    type: event.msgType || 'text',
    createdAt: now
  })

  // Update conversation: last message + unread counts for all recipients
  const recipients = members.filter(id => id !== OPENID)

  const updateData = {
    lastMessage: { content: content.slice(0, 50), createdAt: now },
    updatedAt: now
  }
  // Add recipients to unreadOpenids and increment their unread counts
  recipients.forEach(r => {
    updateData[`unreadCounts.${r}`] = _.inc(1)
  })

  // Rebuild unreadOpenids to include all recipients (remove duplicates)
  const existingUnread = conversation.unreadOpenids || []
  const newUnread = [...new Set([...existingUnread, ...recipients])]
  updateData.unreadOpenids = newUnread

  await db.collection(CONVERSATIONS_COLLECTION).doc(convId).update({ data: updateData })
  await trimConversationMessages(convId)
  const senderRes = await db.collection('users').where({ openid: OPENID }).limit(1).get().catch(() => ({ data: [] }))
  await sendChatNotifications({
    recipients,
    conversationId: convId,
    senderName: senderRes.data[0]?.nickname || conversation.title || '汪友',
    content,
    type: event.msgType || 'text',
    createdAt: now
  })

  return { messageId: msgResult._id, conversationId: convId }
}