rebreak-monorepo/apps/rebreak-native/hooks/useChatRealtime.ts

import { useEffect } from "react";
import { supabase } from "../lib/supabase";
import type { RealtimeChannel } from "@supabase/supabase-js";

/**
 * Realtime-Subscription für DM-Konversation:
 * Lauscht auf INSERT in rebreak.direct_messages mit sender_id=eq.{partnerId}.
 * Filter: Wir bekommen nur Nachrichten DES Partners (eigene werden lokal optimistisch
 * hinzugefügt). callback erhält die rohe Postgres-Row.
 */
export function useDmRealtime(
  partnerId: string | undefined,
  onInsert: (row: any) => void,
  enabled: boolean = true,
  // Partner UPDATEt eine eigene Message (z.B. Soft-Delete deleted_at, read_at).
  onUpdate?: (row: any) => void,
  // Irgendeine DM-Reaktion hat sich geändert → Caller refetcht die Konversation.
  // (Die direct_message_reactions-Tabelle hat keine Partner-Spalte zum Filtern;
  //  für DM-Volumen ist ein Refetch-on-any akzeptabel.)
  onReactionChange?: () => void,
) {
  useEffect(() => {
    if (!enabled || !partnerId) return;
    let channel: RealtimeChannel | null = null;
    let cancelled = false;
    let reconnectTimer: ReturnType<typeof setTimeout> | null = null;

    async function subscribe() {
      const { data } = await supabase.auth.getSession();
      if (cancelled || !data.session?.access_token) return;

      channel = supabase
        .channel(`dm:${partnerId}:${Date.now()}`)
        .on(
          "postgres_changes",
          {
            event: "INSERT",
            schema: "rebreak",
            table: "direct_messages",
            filter: `sender_id=eq.${partnerId}`,
          },
          (payload: any) => {
            onInsert(payload.new);
          },
        )
        .on(
          "postgres_changes",
          {
            event: "UPDATE",
            schema: "rebreak",
            table: "direct_messages",
            filter: `sender_id=eq.${partnerId}`,
          },
          (payload: any) => {
            onUpdate?.(payload.new);
          },
        )
        .on(
          "postgres_changes",
          {
            event: "*",
            schema: "rebreak",
            table: "direct_message_reactions",
          },
          () => {
            onReactionChange?.();
          },
        )
        .subscribe((status) => {
          if (status === "CHANNEL_ERROR" || status === "TIMED_OUT") {
            cleanup();
            if (reconnectTimer) clearTimeout(reconnectTimer);
            reconnectTimer = setTimeout(() => {
              if (!cancelled) subscribe();
            }, 3000);
          }
        });
    }

    function cleanup() {
      if (channel) {
        supabase.removeChannel(channel);
        channel = null;
      }
    }

    subscribe();
    return () => {
      cancelled = true;
      if (reconnectTimer) clearTimeout(reconnectTimer);
      cleanup();
    };
  }, [partnerId, enabled, onInsert, onUpdate, onReactionChange]);
}

/**
 * Realtime für Gruppen-Chat: lauscht auf INSERT in rebreak.chat_messages mit room_id=eq.{roomId}.
 */
export function useRoomRealtime(
  roomId: string | undefined,
  myUserId: string | undefined,
  onInsert: (row: any) => void,
  enabled: boolean = true,
) {
  useEffect(() => {
    if (!enabled || !roomId) return;
    let channel: RealtimeChannel | null = null;
    let cancelled = false;
    let reconnectTimer: ReturnType<typeof setTimeout> | null = null;

    async function subscribe() {
      const { data } = await supabase.auth.getSession();
      if (cancelled || !data.session?.access_token) return;

      channel = supabase
        .channel(`room:${roomId}:${Date.now()}`)
        .on(
          "postgres_changes",
          {
            event: "INSERT",
            schema: "rebreak",
            table: "chat_messages",
            filter: `room_id=eq.${roomId}`,
          },
          (payload: any) => {
            // Eigene Nachrichten überspringen (lokal optimistisch hinzugefügt)
            if (payload.new?.user_id === myUserId) return;
            onInsert(payload.new);
          },
        )
        .subscribe((status) => {
          if (status === "CHANNEL_ERROR" || status === "TIMED_OUT") {
            cleanup();
            if (reconnectTimer) clearTimeout(reconnectTimer);
            reconnectTimer = setTimeout(() => {
              if (!cancelled) subscribe();
            }, 3000);
          }
        });
    }

    function cleanup() {
      if (channel) {
        supabase.removeChannel(channel);
        channel = null;
      }
    }

    subscribe();
    return () => {
      cancelled = true;
      if (reconnectTimer) clearTimeout(reconnectTimer);
      cleanup();
    };
  }, [roomId, myUserId, enabled, onInsert]);
}