pearcord-platform/mixins/domain/workers-mixin.js

'use strict'

const workersMixin = {
  _workerGossipKeys: null,
  _workerRegistryHealWatermark: null,
  _workerTrialHealWatermark: null,

  _initWorkersMixinState () {
    if (!this._workerGossipKeys) {
      this._workerGossipKeys = new Set()
    }
  },

  _shouldGossipWorkerRelease (payload) {
    this._initWorkersMixinState()
    if (!payload?.guildId || !payload?.workerVersion) return true
    const hash = `${payload.guildId}:${payload.workerVersion}:${payload.channelTag || 'stable'}:${payload.publishedAt || 0}:${payload.pearUri || ''}`
    if (this._workerGossipKeys.has(hash)) return false
    this._workerGossipKeys.add(hash)
    if (this._workerGossipKeys.size > 8192) {
      const first = this._workerGossipKeys.values().next().value
      if (first) this._workerGossipKeys.delete(first)
    }
    return true
  },

  _shouldGossipWorkerTrial (result) {
    this._initWorkersMixinState()
    if (!result?.guildId || !result?.templateId) return true
    const hash = `${result.guildId}:${result.templateId}:${result.ok ? 1 : 0}:${result.eventsSeen || 0}:${result.receiptsSeen || 0}:${result.holdMs || 0}`
    if (this._workerGossipKeys.has(hash)) return false
    this._workerGossipKeys.add(hash)
    if (this._workerGossipKeys.size > 8192) {
      const first = this._workerGossipKeys.values().next().value
      if (first) this._workerGossipKeys.delete(first)
    }
    return true
  },

  async _healWorkerRegistryOnPartition (guildId) {
    const gid = guildId || this.guild?.guild?.id || null
    const span = this.log.time('integration.worker', {
      spanKind: 'integration.worker',
      guildId: gid,
      context: 'heal.registry'
    })
    try {
      if (!gid || !this.listIntegrationWorkerTemplates) {
        span.end({ relisted: 0, skipped: true, guildWorkerCount: 0 })
        return { relisted: 0, skipped: true }
      }
      const rows = this.listIntegrationWorkerTemplates() || []
      let relisted = 0
      for (const row of rows) {
        if (this._shouldGossipWorkerTrial({ guildId: gid, templateId: row.id, ok: true, eventsSeen: 0 })) {
          relisted++
        }
      }
      const watermark = Date.now()
      this._workerRegistryHealWatermark = watermark
      span.end({ relisted, watermark, guildWorkerCount: rows.length, bridgeKind: 'integration.worker' })
      return { relisted, watermark, guildWorkerCount: rows.length }
    } catch (err) {
      this.log.error('integration.worker error', {
        guildId: gid,
        context: 'heal.registry',
        error: err?.message || String(err)
      })
      span.fail(err)
      return { relisted: 0, error: err?.message || String(err) }
    }
  },

  async _healWorkerTrialCursorOnPartition (guildId) {
    const gid = guildId || this.guild?.guild?.id || null
    const span = this.log.time('integration.worker', {
      spanKind: 'integration.worker',
      guildId: gid,
      context: 'heal.trial'
    })
    try {
      if (!gid) {
        span.end({ relisted: 0, skipped: true, guildWorkerCount: 0 })
        return { relisted: 0, skipped: true }
      }
      const list = this.listIntegrationWorkerTemplates ? this.listIntegrationWorkerTemplates() : []
      const watermark = Date.now()
      this._workerTrialHealWatermark = watermark
      span.end({
        relisted: 0,
        watermark,
        guildWorkerCount: list.length,
        guildCount: (this.guilds || []).length,
        activeChannelId: this.activeChannelId || null,
        bridgeKind: 'integration.worker'
      })
      return { relisted: 0, watermark, guildWorkerCount: list.length }
    } catch (err) {
      this.log.error('integration.worker error', {
        guildId: gid,
        context: 'heal.trial',
        error: err?.message || String(err)
      })
      span.fail(err)
      return { relisted: 0, error: err?.message || String(err) }
    }
  }
}

module.exports = { workersMixin }