app/services/sse/sseSource.ts

import { EventSourcePolyfill } from 'event-source-polyfill'
import type { MercureEntityUpdate } from '~/types/interfaces'

class SseSource {
  protected readonly url: URL
  protected readonly onOpen: () => void
  protected readonly onMessage: (eventData: MercureEntityUpdate) => void
  protected readonly onClose: () => void
  protected readonly withCredentials: boolean
  protected eventSource: EventSourcePolyfill | null = null
  constructor(
    mercureHubBaseUrl: string,
    topic: string,
    onOpen: () => void,
    onMessage: (eventData: MercureEntityUpdate) => void,
    onClose: () => void,
    withCredentials: boolean = true,
  ) {
    this.url = new URL(mercureHubBaseUrl)
    this.url.searchParams.append('topic', topic)
    this.onOpen = onOpen
    this.onMessage = onMessage
    this.onClose = onClose
    this.withCredentials = withCredentials
  }

  protected createEventSource(
    url: string,
    withCredentials: boolean,
  ): EventSourcePolyfill {
    return new EventSourcePolyfill(url, {
      withCredentials,
      heartbeatTimeout: 45 * 1000, // in ms
    })
  }

  public isConnected() {
    return (
      this.eventSource !== null &&
      this.eventSource.readyState === EventSourcePolyfill.OPEN
    )
  }

  public subscribe() {
    if (this.isConnected()) {
      throw new Error('SSE - Already subscribed to this event source')
    }
    if (import.meta.server) {
      throw new Error('SSE - Cannot subscribe on server side')
    }

    this.eventSource = this.createEventSource(
      this.url.toString(),
      this.withCredentials,
    )

    this.eventSource.onerror = (event) => {
      if (event.target.readyState === EventSourcePolyfill.CLOSED) {
        console.log('SSE closed')
      }

      this.eventSource!.close()
    }
    this.eventSource.onopen = () => {
      console.log('SSE - Listening for events...')
      this.onOpen()
    }
    this.eventSource.onmessage = (event) => {
      const data = JSON.parse(event.data)
      this.onMessage(data)
    }
  }

  public unsubscribe() {
    if (
      this.eventSource === null ||
      this.eventSource.readyState === EventSourcePolyfill.CLOSED
    ) {
      return
    }
    this.eventSource.close()
    this.onClose()
    console.log('SSE - Subscription closed')
  }
}

export default SseSource