akmon/utils/gatewayRealtimeService.uts

import supa, { supaReady } from '@/components/supadb/aksupainstance.uts'
import AkSupaRealtime from '@/components/supadb/aksuparealtime.uts'
import { WS_URL, SUPA_KEY } from '@/ak/config.uts'

export type GatewayNode = {
  id: string
  name: string
  mqtt_client_id: string
  version: string | null
  region: string | null
  tags: string | null
  updated_at: string
}

export type GatewayRealtimeHandlers = {
  onInitial: (nodes: GatewayNode[]) => void
  onInsert: (node: GatewayNode) => void
  onUpdate: (node: GatewayNode) => void
  onDelete: (id: string) => void
  onError: (err: any) => void
}

export type GatewayRealtimeSubscription = {
  dispose: () => void
}

type SubscriptionGuard = {
  active: boolean
}

type ActiveSubscription = {
  handlers: GatewayRealtimeHandlers
  guard: SubscriptionGuard
}

async function ensureSupaSession(): Promise<void> {
  try {
    await supaReady
  } catch (err) {
    console.error('Supabase session initialization error', err)
  }
}

function toGatewayNode(row: UTSJSONObject): GatewayNode | null {
  try {
    const id = row.getString('id')
    const name = row.getString('name')
    const mqttClientId = row.getString('mqtt_client_id')
    const updatedAt = row.getString('updated_at')
    
    if (id == null || name == null || mqttClientId == null || updatedAt == null) {
      return null
    }

    return {
      id: id,
      name: name,
      mqtt_client_id: mqttClientId,
      version: row.getString('version'),
      region: row.getString('region'),
      tags: JSON.stringify(row.get('tags')), // Simple stringify for display
      updated_at: updatedAt
    }
  } catch (e) {
    console.error('toGatewayNode error', e)
    return null
  }
}

function extractRecord(payload: any): UTSJSONObject | null {
  try {
    if (payload == null) return null
    const wrapper = payload as UTSJSONObject
    const directNew = wrapper.get('new') as UTSJSONObject | null
    if (directNew != null) return directNew
    const record = wrapper.get('record') as UTSJSONObject | null
    if (record != null) return record
    return null
  } catch (err) {
    console.error('extractRecord error', err)
  }
  return null
}

function extractOldRecord(payload: any): UTSJSONObject | null {
  try {
    if (payload == null) return null
    const wrapper = payload as UTSJSONObject
    const directOld = wrapper.get('old') as UTSJSONObject | null
    if (directOld != null) return directOld
    const oldRecord = wrapper.get('old_record') as UTSJSONObject | null
    if (oldRecord != null) return oldRecord
    return null
  } catch (err) {
    console.error('extractOldRecord error', err)
  }
  return null
}

export class GatewayRealtimeService {
  private static rt: AkSupaRealtime | null = null
  private static initializing: Promise<void> | null = null
  private static initialized: boolean = false
  private static subscriptions: Array<ActiveSubscription> = []
  private static reconnectAttempts: number = 0
  private static reconnectTimer: number | null = null
  private static reconnecting: boolean = false

  private static async waitForRealtimeOpen(timeoutMs: number = 5000): Promise<void> {
    const start = Date.now()
    while (true) {
      if (this.rt != null && this.rt.isOpen == true) {
        return
      }
      if (Date.now() - start > timeoutMs) {
        throw new Error('Realtime socket not ready')
      }
      await new Promise<void>((resolve) => {
        setTimeout(() => resolve(), 50)
      })
    }
  }

  private static clearReconnectTimer(): void {
    const timer = this.reconnectTimer
    if (timer != null) {
      clearTimeout(timer)
      this.reconnectTimer = null
    }
  }

  private static handleSocketClose(origin: string | null): void {
    this.initialized = false
    this.rt = null
    if (this.subscriptions.length == 0) {
      this.clearReconnectTimer()
      this.reconnectAttempts = 0
      return
    }
    if (this.reconnecting == true) {
      return
    }
    console.warn('[GatewayRealtimeService] realtime closed, scheduling reconnect', origin ?? 'unknown')
    this.scheduleReconnect()
  }

  private static async scheduleReconnect(): Promise<void> {
    if (this.reconnectTimer != null) return
    if (this.subscriptions.every((sub) => sub.guard.active !== true)) {
      this.subscriptions = []
      this.reconnectAttempts = 0
      return
    }
    const attempt = this.reconnectAttempts + 1
    this.reconnectAttempts = attempt
    const baseDelay = Math.min(Math.pow(2, attempt - 1) * 1000, 10000)
    const jitter = Math.floor(Math.random() * 200)
    const delay = baseDelay + jitter
    console.log('[GatewayRealtimeService] reconnect scheduled in', delay, 'ms (attempt', attempt, ')')
    this.reconnectTimer = setTimeout(() => {
      const runner = async () => {
        this.reconnectTimer = null
        if (this.subscriptions.every((sub) => sub.guard.active !== true)) {
          this.subscriptions = []
          this.reconnectAttempts = 0
          return
        }
        this.reconnecting = true
        try {
          await this.ensureRealtime()
          await this.waitForRealtimeOpen(3000)
          this.reconnectAttempts = 0
          await this.resubscribeAll()
        } catch (err) {
          console.error('reconnect attempt failed', err)
          this.scheduleReconnect()
        } finally {
          this.reconnecting = false
        }
      }
      runner().catch((err) => {
        console.error('reconnect timer runner error', err)
      })
    }, delay)
  }

  private static async resubscribeAll(): Promise<void> {
    let hadFailure = false
    for (let i = 0; i < this.subscriptions.length; i++) {
      const record = this.subscriptions[i]
      if (record.guard.active !== true) continue
      try {
        await this.performSubscription(record)
      } catch (err) {
        console.error('resubscribe error', err)
        record.handlers?.onError?.(err)
        hadFailure = true
      }
    }
    if (hadFailure) {
      this.scheduleReconnect()
    }
  }

  private static removeSubscription(record: ActiveSubscription): void {
    const index = this.subscriptions.indexOf(record)
    if (index >= 0) {
      this.subscriptions.splice(index, 1)
    }
    if (this.subscriptions.length == 0) {
      this.clearReconnectTimer()
      this.reconnectAttempts = 0
    }
  }

  private static async performSubscription(record: ActiveSubscription): Promise<void> {
    if (record.guard.active !== true) return
    const handlers = record.handlers
    const maxAttempts = 2
    let attempt = 0
    
    // 1. Fetch Initial Data
    try {
      await ensureSupaSession()
      const { data, error } = await supa.from('chat_gateway_nodes').select('*').order('name', { ascending: true })
      if (error != null) {
        throw new Error(error.message)
      }
      const nodes: GatewayNode[] = []
      if (data != null) {
        const list = data as UTSArray<UTSJSONObject>
        for (let i = 0; i < list.length; i++) {
          const node = toGatewayNode(list[i])
          if (node != null) {
            nodes.push(node)
          }
        }
      }
      if (record.guard.active) {
        handlers.onInitial(nodes)
      }
    } catch (fetchErr) {
      console.error('Initial fetch failed', fetchErr)
      handlers.onError(fetchErr)
    }

    // 2. Subscribe to Realtime
    while (attempt < maxAttempts) {
      attempt += 1
      if (record.guard.active !== true) return
      try {
        if (this.rt == null || this.rt.isOpen !== true) {
          await this.ensureRealtime()
          await this.waitForRealtimeOpen(2000)
        }
        const realtime = this.rt
        if (realtime == null) {
          throw new Error('Realtime client not initialized')
        }
        console.log('[GatewayRealtimeService] subscribing to chat_gateway_nodes')
        
        realtime.subscribePostgresChanges({
          event: '*',
          schema: 'public',
          table: 'chat_gateway_nodes',
          filter: 'mqtt_client_id=neq.null',
          topic: 'realtime:gateway_nodes',
          onChange: (payload: any) => {
            if (record.guard.active !== true) return
            const wrapper = payload as UTSJSONObject
            let eventType = wrapper.getString('type')
            if (eventType == null) {
               eventType = wrapper.getString('eventType')
            }
            
            if (eventType == 'INSERT') {
              const row = extractRecord(payload)
              if (row != null) {
                const node = toGatewayNode(row)
                if (node != null) handlers.onInsert(node)
              }
            } else if (eventType == 'UPDATE') {
              const row = extractRecord(payload)
              if (row != null) {
                const node = toGatewayNode(row)
                if (node != null) handlers.onUpdate(node)
              }
            } else if (eventType == 'DELETE') {
              const oldRow = extractOldRecord(payload)
              if (oldRow != null) {
                const id = oldRow.getString('id')
                if (id != null) handlers.onDelete(id)
              }
            }
          }
        })
        console.log('[GatewayRealtimeService] subscription ready')
        return
      } catch (subscribeErr) {
        console.error('performSubscription error', subscribeErr, 'attempt', attempt)
        if (attempt >= maxAttempts) {
          throw subscribeErr
        }
        await new Promise<void>((resolve) => {
          setTimeout(() => resolve(), 200)
        })
      }
    }
  }

  private static async ensureRealtime(): Promise<void> {
    if (this.rt != null && this.rt.isOpen == true) {
      this.initialized = true
      return
    }
    if (this.initializing != null) {
      await this.initializing
      return
    }
    await ensureSupaSession()
    let token: string | null = null
    try {
      const session = supa.getSession()
      if (session != null) {
        token = session.session?.access_token ?? null
      }
    } catch (_) { }

    let resolveReady: ((value: void) => void) | null = null
    let rejectReady: ((reason: any) => void) | null = null
    const readyPromise = new Promise<void>((resolve, reject) => {
      resolveReady = resolve as (value: void) => void
      rejectReady = reject
    })

    this.initializing = readyPromise
    const realtime = new AkSupaRealtime({
      url: WS_URL,
      channel: 'realtime:gateway_nodes',
      apikey: SUPA_KEY,
      token: token,
      onMessage: (_data: any) => { },
      onOpen: (_res: any) => {
        this.initialized = true
        this.reconnectAttempts = 0
        this.clearReconnectTimer()
        this.reconnecting = false
        const resolver = resolveReady
        resolveReady = null
        rejectReady = null
        this.initializing = null
        resolver?.()
      },
      onError: (err: any) => {
        if (resolveReady != null) {
          const rejector = rejectReady
          resolveReady = null
          rejectReady = null
          this.initializing = null
          rejector?.(err)
        }
      },
      onClose: (_res: any) => {
        if (resolveReady != null) {
          const rejector = rejectReady
          resolveReady = null
          rejectReady = null
          this.initializing = null
          rejector?.(new Error('Realtime connection closed before ready'))
        } else {
          this.handleSocketClose('close')
        }
      }
    })
    this.rt = realtime
    this.rt?.connect()
    try {
      await readyPromise
    } finally {
      this.initializing = null
    }
  }

  static async subscribeGateways(handlers: GatewayRealtimeHandlers): Promise<GatewayRealtimeSubscription> {
    const guard: SubscriptionGuard = { active: true }
    const record: ActiveSubscription = {
      handlers: handlers,
      guard: guard
    }
    this.subscriptions.push(record)
    try {
      await this.performSubscription(record)
    } catch (err) {
      console.error('subscribeGateways failed', err)
      guard.active = false
      this.removeSubscription(record)
      handlers.onError(err)
      return { dispose: () => { } }
    }
    return {
      dispose: () => {
        if (guard.active !== true) return
        guard.active = false
        this.removeSubscription(record)
      }
    }
  }

  static closeRealtime(): void {
    try {
      if (this.rt != null) {
        this.rt.close({ code: 1000, reason: 'manual close' })
      }
    } catch (err) {
      console.error('closeRealtime error', err)
    } finally {
      this.rt = null
      this.initialized = false
      this.initializing = null
      this.reconnecting = false
      this.clearReconnectTimer()
      this.reconnectAttempts = 0
    }
  }
}

export default GatewayRealtimeService