pearcord-platform/mark-read-fanout.js

'use strict'

const DEFAULT_DEBOUNCE_MS = 80
const DEFAULT_WATCHDOG_MS = 5000

/**
 * Coalesce rapid mark-channel-read calls during fast channel/thread switches (P408-16).
 */
class MarkReadFanoutQueue {
  constructor ({ log, onFlush }) {
    this.log = log
    this.onFlush = onFlush
    /** @type {Map<string, { channelId: string, guildId: string, parentId?: string|null, enqueuedAt: number }>} */
    this.pending = new Map()
    this._timer = null
    this._inFlight = null
    this._flushStartedAt = 0
    this._watchdogTimer = null
  }

  enqueue ({ channelId, guildId, parentId = null }) {
    if (!channelId || !guildId) return
    const key = `${guildId}:${channelId}`
    this.pending.set(key, {
      channelId,
      guildId,
      parentId,
      enqueuedAt: Date.now()
    })
    if (parentId) {
      const parentKey = `${guildId}:${parentId}`
      if (!this.pending.has(parentKey)) {
        this.pending.set(parentKey, {
          channelId: parentId,
          guildId,
          parentId: null,
          enqueuedAt: Date.now()
        })
      }
    }
    this._scheduleFlush()
  }

  _scheduleFlush () {
    if (this._timer) return
    const ms = Number(process.env.PEARCORD_MARK_READ_FANOUT_MS) || DEFAULT_DEBOUNCE_MS
    this._timer = setTimeout(() => {
      this._timer = null
      void this._flush()
    }, ms)
  }

  _armWatchdog () {
    if (this._watchdogTimer) return
    const ms = Number(process.env.PEARCORD_MARK_READ_FANOUT_WATCHDOG_MS) || DEFAULT_WATCHDOG_MS
    this._watchdogTimer = setTimeout(() => {
      this._watchdogTimer = null
      if (this._inFlight && this._flushStartedAt) {
        const stalledMs = Date.now() - this._flushStartedAt
        this.log.warn('inbox.mark-read-fanout stalled', {
          spanKind: 'inbox.mark-read-fanout',
          stalledMs,
          pending: this.pending.size
        })
      }
    }, ms)
  }

  async _flush () {
    if (this._inFlight) return this._inFlight
    const batch = [...this.pending.values()]
    this.pending.clear()
    if (!batch.length) return
    this._flushStartedAt = Date.now()
    this._armWatchdog()
    this._inFlight = (async () => {
      for (const row of batch) {
        await this.onFlush(row.channelId, row.guildId)
      }
    })()
      .catch((err) => {
        this.log.error('inbox.mark-read-fanout error', {
          error: err?.message || String(err),
          batchSize: batch.length
        })
      })
      .finally(() => {
        this._inFlight = null
        this._flushStartedAt = 0
        if (this._watchdogTimer) {
          clearTimeout(this._watchdogTimer)
          this._watchdogTimer = null
        }
        if (this.pending.size) this._scheduleFlush()
      })
    return this._inFlight
  }
}

module.exports = { MarkReadFanoutQueue }