141 lines
5.1 KiB
JavaScript
141 lines
5.1 KiB
JavaScript
import { onMounted, onUnmounted } from 'vue'
|
||
import { useContributionPolling } from './useContributionPolling.js'
|
||
import { getActivitySocket } from '@/utils/socket/ActivitySocket.js'
|
||
|
||
/**
|
||
* 连击合并常量与工具函数 —— WS 与轮询两条路径共用
|
||
* - 同一 user_id+item_id 在 COMBO_WINDOW_MS 内的多条 → 合并为 1 条
|
||
* - 合并后 id = 同组最早 id（与后端 first_id 一致，前端按 id 去重不会双计）
|
||
* - quantity = 同组 quantity 之和
|
||
*/
|
||
const COMBO_WINDOW_MS = 3000
|
||
|
||
export function mergeComboRecords(records) {
|
||
if (!Array.isArray(records) || records.length === 0) return records
|
||
|
||
// 倒序遍历：新→旧；同 key 在窗口内合并到索引更大的（更新的）那条
|
||
const result = [...records]
|
||
const indexByKey = new Map() // key = `${user_id}:${item_id}` -> result 中索引
|
||
|
||
for (let i = result.length - 1; i >= 0; i--) {
|
||
const r = result[i]
|
||
const key = `${r.user_id}:${r.item_id}`
|
||
const existingIdx = indexByKey.get(key)
|
||
if (existingIdx !== undefined) {
|
||
const existing = result[existingIdx]
|
||
if (Math.abs(existing.created_at - r.created_at) <= COMBO_WINDOW_MS) {
|
||
existing.quantity += r.quantity
|
||
existing.combo_count = existing.quantity
|
||
if (r.id < existing.id) existing.id = r.id // 取最早 id
|
||
result.splice(i, 1)
|
||
continue
|
||
}
|
||
}
|
||
indexByKey.set(key, i)
|
||
}
|
||
return result
|
||
}
|
||
|
||
/**
|
||
* 贡献实时推送 composable（WS 优先，断线降级为轮询）
|
||
* @param {import('vue').Ref<string|number>} activityId
|
||
* @param {import('vue').Ref<boolean>} isPageActive
|
||
*/
|
||
export function useContributionRealtime(activityId, isPageActive) {
|
||
const MAX_RECORDS = 5
|
||
|
||
const {
|
||
records,
|
||
visible,
|
||
loading,
|
||
error,
|
||
start: startPolling,
|
||
stop: stopPolling,
|
||
reset: resetPolling,
|
||
highestIdRef,
|
||
} = useContributionPolling(activityId, isPageActive)
|
||
|
||
const socket = getActivitySocket()
|
||
let usingWS = false
|
||
|
||
function onWsMessage(payload) {
|
||
if (!payload || Number(payload.activity_id) !== Number(activityId.value)) return
|
||
if (!payload.record) return
|
||
const record = payload.record
|
||
if (record.id > highestIdRef()) {
|
||
// 追加到列表末尾（与轮询的方向不同），保留最近 MAX_RECORDS 条
|
||
records.value = mergeComboRecords([...records.value, record]).slice(-MAX_RECORDS)
|
||
}
|
||
}
|
||
|
||
function onWsConnect() {
|
||
if (usingWS) return
|
||
// 即使 WS 已连接，如果后端 Pub/Sub 不可用，也不切到 WS，保持轮询
|
||
if (!socket.isPubSubEnabled()) {
|
||
console.log('[useContributionRealtime] WS connected but pubsub disabled, keeping polling')
|
||
return
|
||
}
|
||
usingWS = true
|
||
stopPolling() // 停掉可能的轮询
|
||
socket.subscribe(activityId.value, ['contributions'])
|
||
}
|
||
|
||
function onWsDisconnect() {
|
||
if (!usingWS) return
|
||
usingWS = false
|
||
startPolling() // 降级为轮询
|
||
}
|
||
|
||
// 后端上报 pub/sub 不可用：如果之前已切到 WS，回退到轮询
|
||
function onPubsubDisabled() {
|
||
if (!usingWS) return
|
||
console.warn('[useContributionRealtime] Pub/Sub disabled, falling back to polling')
|
||
usingWS = false
|
||
socket.unsubscribe(activityId.value, ['contributions'])
|
||
startPolling()
|
||
}
|
||
|
||
socket.onContributionsResponse(onWsMessage)
|
||
socket.on('connect', onWsConnect)
|
||
socket.on('disconnect', onWsDisconnect)
|
||
socket.on('pubsub_disabled', onPubsubDisabled)
|
||
|
||
onMounted(() => {
|
||
// 总是调用 connect():SocketManager.connect() 内部会判断 token 是否变化
|
||
// (详见 useMessageRealtime.js 注释)。这样能确保用户切换登录后,
|
||
// contribution channel 也能用新 token 重新订阅,而不是复用上一个用户的 WS。
|
||
const token = uni.getStorageSync('access_token')
|
||
if (token) {
|
||
socket.connect(token).catch(err => console.warn('[useContributionRealtime] connect error:', err))
|
||
}
|
||
// 同步分支:如果 WS 已连接（单例复用导致 'connect' 事件不会再次触发）且 pubsub 可用,
|
||
// 必须直接调 onWsConnect 停轮询,否则 polling 会一直跑。
|
||
// 异步分支:WS 还没连上时,先起 polling 兜底;
|
||
// 等 'connect' 事件触发 onWsConnect 后会停掉轮询。
|
||
// 但如果 pubsub 被禁用,即使 WS 已连接也不切,保持轮询。
|
||
if (socket.isConnected && socket.isPubSubEnabled()) {
|
||
onWsConnect()
|
||
} else {
|
||
startPolling()
|
||
}
|
||
})
|
||
|
||
onUnmounted(() => {
|
||
if (usingWS) socket.unsubscribe(activityId.value, ['contributions'])
|
||
socket.off('connect', onWsConnect)
|
||
socket.off('disconnect', onWsDisconnect)
|
||
socket.off('pubsub_disabled', onPubsubDisabled)
|
||
socket.offContributionsResponse(onWsMessage)
|
||
stopPolling()
|
||
resetPolling()
|
||
})
|
||
|
||
return {
|
||
records,
|
||
visible,
|
||
loading,
|
||
error,
|
||
isUsingWS: () => usingWS,
|
||
}
|
||
}
|