useActivitySubscription.ts

'use client';

import { useCallback, useEffect, useRef, useState } from 'react';
import { useQueryClient } from '@tanstack/react-query';
import { getQueryKey } from '@trpc/react-query';
import { eventsTailStore } from '@/lib/telemetry/eventsTail';
import { trpc } from '@/lib/trpc';
import type { ActivityEvent } from '@/server/events/types';
import { applyEvent, type ActivityResponse } from './applyEvent';
import {
  LEADER_LOCK_TTL_MS,
  isLockHeldBy,
  refreshLock,
  releaseLock,
  tryAcquire,
} from './leaderElection';
import { getOptimisticStore } from './optimisticStore';

const BROADCAST_CHANNEL = 'super-activity-sse';
const BACKOFF_MS = [1_000, 2_000, 5_000, 10_000, 10_000];
const LEADER_HEARTBEAT_MS = 1_000;
const LEADER_POLL_MS = 750;
const MAX_LAST_EVENT_ID = 'super-activity-last-event-id';

export type SubscriptionStatus = 'connecting' | 'open' | 'reconnecting' | 'offline';

export type SubscriptionRole = 'leader' | 'follower';

export type ActivitySubscriptionState = {
  status: SubscriptionStatus;
  role: SubscriptionRole | null;
  lastEventAt: number | null;
  eventsReceived: number;
};

type BroadcastMessage =
  | { kind: 'event'; event: ActivityEvent }
  | { kind: 'status'; status: SubscriptionStatus; lastEventAt: number | null };

function createTabId(): string {
  if (typeof crypto !== 'undefined' && 'randomUUID' in crypto) {
    return crypto.randomUUID();
  }
  return `tab-${Math.random().toString(36).slice(2)}-${Date.now()}`;
}

function loadLastEventId(): number {
  if (typeof sessionStorage === 'undefined') return 0;
  try {
    const raw = sessionStorage.getItem(MAX_LAST_EVENT_ID);
    if (!raw) return 0;
    const parsed = Number.parseInt(raw, 10);
    return Number.isFinite(parsed) && parsed >= 0 ? parsed : 0;
  } catch {
    return 0;
  }
}

function saveLastEventId(seq: number): void {
  if (typeof sessionStorage === 'undefined') return;
  try {
    sessionStorage.setItem(MAX_LAST_EVENT_ID, String(seq));
  } catch {
    // ignore
  }
}

export function useActivitySubscription(): ActivitySubscriptionState {
  const queryClient = useQueryClient();
  const [state, setState] = useState<ActivitySubscriptionState>({
    status: 'connecting',
    role: null,
    lastEventAt: null,
    eventsReceived: 0,
  });

  const tabIdRef = useRef<string | null>(null);
  if (tabIdRef.current === null) {
    tabIdRef.current = createTabId();
  }

  const applyToCache = useCallback(
    (event: ActivityEvent) => {
      const queryKey = getQueryKey(
        trpc.deals.getActivityOfferPlacements,
        { device_platform: 'web' },
        'query',
      );
      queryClient.setQueriesData<ActivityResponse>({ queryKey }, (current) => {
        if (!current) return current;
        return applyEvent(current, event);
      });
      getOptimisticStore().resolveEvent(event);
      saveLastEventId(event.seq);
      eventsTailStore.push(event);
    },
    [queryClient],
  );

  useEffect(() => {
    if (typeof window === 'undefined') return;
    const tabId = tabIdRef.current!;

    let eventSource: EventSource | null = null;
    let channel: BroadcastChannel | null = null;
    let heartbeat: ReturnType<typeof setInterval> | null = null;
    let pollInterval: ReturnType<typeof setInterval> | null = null;
    let backoffTimer: ReturnType<typeof setTimeout> | null = null;
    let backoffIndex = 0;
    let disposed = false;
    let currentRole: SubscriptionRole | null = null;
    let lastStatus: SubscriptionStatus = 'connecting';

    const supportsBroadcast = typeof BroadcastChannel !== 'undefined';
    if (supportsBroadcast) {
      channel = new BroadcastChannel(BROADCAST_CHANNEL);
    }

    const updateState = (patch: Partial<ActivitySubscriptionState>) => {
      setState((prev) => {
        const next = { ...prev, ...patch };
        if (
          prev.status === next.status &&
          prev.role === next.role &&
          prev.lastEventAt === next.lastEventAt &&
          prev.eventsReceived === next.eventsReceived
        ) {
          return prev;
        }
        return next;
      });
    };

    const broadcastStatus = (status: SubscriptionStatus, lastEventAt: number | null) => {
      if (!channel) return;
      if (currentRole !== 'leader') return;
      const message: BroadcastMessage = { kind: 'status', status, lastEventAt };
      channel.postMessage(message);
    };

    const handleEvent = (event: ActivityEvent) => {
      applyToCache(event);
      const eventTime = Date.now();
      setState((prev) => {
        const nextStatus: SubscriptionStatus =
          prev.status === 'offline' ? 'offline' : 'open';
        if (nextStatus !== lastStatus && currentRole === 'leader') {
          lastStatus = nextStatus;
        }
        return {
          ...prev,
          status: nextStatus,
          lastEventAt: eventTime,
          eventsReceived: prev.eventsReceived + 1,
        };
      });
      if (channel && currentRole === 'leader') {
        const message: BroadcastMessage = { kind: 'event', event };
        channel.postMessage(message);
      }
    };

    const clearHeartbeat = () => {
      if (heartbeat) {
        clearInterval(heartbeat);
        heartbeat = null;
      }
    };

    const scheduleReconnect = () => {
      if (disposed) return;
      const delay = BACKOFF_MS[Math.min(backoffIndex, BACKOFF_MS.length - 1)];
      backoffIndex += 1;
      lastStatus = 'reconnecting';
      updateState({ status: 'reconnecting' });
      broadcastStatus('reconnecting', null);
      backoffTimer = setTimeout(() => {
        backoffTimer = null;
        if (disposed) return;
        openLeaderStream();
      }, delay);
    };

    const openLeaderStream = () => {
      if (disposed) return;
      if (eventSource) {
        eventSource.close();
        eventSource = null;
      }
      lastStatus = 'connecting';
      updateState({ status: 'connecting', role: 'leader' });
      broadcastStatus('connecting', null);

      try {
        eventSource = new EventSource('/api/events/activity');
      } catch {
        scheduleReconnect();
        return;
      }

      const es = eventSource;

      es.addEventListener('open', () => {
        backoffIndex = 0;
        lastStatus = 'open';
        updateState({ status: 'open' });
        broadcastStatus('open', null);
      });

      es.addEventListener('activity', (ev) => {
        const msg = ev as MessageEvent<string>;
        try {
          const parsed = JSON.parse(msg.data) as ActivityEvent;
          handleEvent(parsed);
        } catch {
          // drop malformed frames silently
        }
      });

      es.addEventListener('snapshot', (ev) => {
        const msg = ev as MessageEvent<string>;
        try {
          const parsed = JSON.parse(msg.data) as { latest_seq: number };
          if (Number.isFinite(parsed.latest_seq)) {
            saveLastEventId(parsed.latest_seq);
          }
        } catch {
          // ignore
        }
      });

      es.addEventListener('error', () => {
        if (disposed) return;
        if (es.readyState === EventSource.CLOSED) {
          scheduleReconnect();
        } else {
          lastStatus = 'reconnecting';
          updateState({ status: 'reconnecting' });
          broadcastStatus('reconnecting', null);
        }
      });
    };

    const becomeLeader = () => {
      if (disposed) return;
      currentRole = 'leader';
      updateState({ role: 'leader' });
      clearHeartbeat();
      heartbeat = setInterval(() => {
        if (disposed) return;
        if (!refreshLock(tabId)) {
          teardownLeader();
          becomeFollower();
          return;
        }
        broadcastStatus(lastStatus, null);
      }, LEADER_HEARTBEAT_MS);
      openLeaderStream();
    };

    const teardownLeader = () => {
      if (eventSource) {
        eventSource.close();
        eventSource = null;
      }
      if (backoffTimer) {
        clearTimeout(backoffTimer);
        backoffTimer = null;
      }
      clearHeartbeat();
      backoffIndex = 0;
      releaseLock(tabId);
    };

    const becomeFollower = () => {
      if (disposed) return;
      currentRole = 'follower';
      updateState({ role: 'follower', status: 'connecting' });
    };

    const handleOffline = () => {
      lastStatus = 'offline';
      updateState({ status: 'offline' });
      if (currentRole === 'leader') {
        broadcastStatus('offline', null);
      }
    };

    const handleOnline = () => {
      if (currentRole === 'leader') {
        if (lastStatus === 'offline') {
          scheduleReconnect();
        }
      }
    };

    const evaluateLeadership = () => {
      if (disposed) return;
      if (currentRole === 'leader') {
        return;
      }
      if (tryAcquire(tabId)) {
        becomeLeader();
      }
    };

    if (channel) {
      channel.addEventListener('message', (ev) => {
        const message = ev.data as BroadcastMessage | undefined;
        if (!message) return;
        if (currentRole === 'leader') return;
        if (message.kind === 'event') {
          handleEvent(message.event);
        } else if (message.kind === 'status') {
          updateState({ status: message.status, lastEventAt: message.lastEventAt ?? null });
        }
      });
    }

    const handlePageHide = () => {
      if (currentRole === 'leader') {
        releaseLock(tabId);
      }
    };

    window.addEventListener('offline', handleOffline);
    window.addEventListener('online', handleOnline);
    window.addEventListener('pagehide', handlePageHide);

    if (navigator && navigator.onLine === false) {
      lastStatus = 'offline';
      updateState({ status: 'offline' });
    }

    if (tryAcquire(tabId)) {
      becomeLeader();
    } else {
      becomeFollower();
    }

    pollInterval = setInterval(() => {
      if (disposed) return;
      if (currentRole === 'leader') {
        if (!isLockHeldBy(tabId)) {
          teardownLeader();
          becomeFollower();
        }
        return;
      }
      evaluateLeadership();
    }, LEADER_POLL_MS);

    return () => {
      disposed = true;
      if (pollInterval) clearInterval(pollInterval);
      teardownLeader();
      window.removeEventListener('offline', handleOffline);
      window.removeEventListener('online', handleOnline);
      window.removeEventListener('pagehide', handlePageHide);
      if (channel) channel.close();
    };
  }, [applyToCache]);

  return state;
}

export { LEADER_LOCK_TTL_MS };