Restore session after channel error

[colinchadwick]

Thanks for the code! I've modified it to restore the session automatically if an error occurs. This usually happens after the computer goes in to standby mode.

import type {
  SupabaseClient,
  RealtimePostgresChangesPayload,
} from '@supabase/supabase-js';

export type Payload = RealtimePostgresChangesPayload<{ [key: string]: any }>;

export interface SupaSnap<T> {
  data: T[];
  payload: Payload;
}

export interface SupaSingleSnap<T> {
  data: T;
  payload: Payload;
}

type SubscribeInput = {
  table: string;
  field?: string;
  value?: string | string[];
  single?: boolean;
  filterName?: FilterNames;
};

// filter types
const _filterNames = ['eq', 'neq', 'gt', 'gte', 'lt', 'lte', 'in'] as const;
type FilterNames = (typeof _filterNames)[number];

type Single<T> = (
  callback: (snap: SupaSingleSnap<T>) => void
) => () => Promise<'error' | 'ok' | 'timed out'>;

export const realtime = <T>(
  supabase: SupabaseClient,
  { schema = 'public', idField = 'id', limit = 100 } = {}
) => {
  let isSubscribed = false;
  const items: any[] = [];

  const _subscribe = ({
    table,
    field,
    value,
    single = false,
    filterName,
  }: SubscribeInput) => {
    const hasFilter = field && value && filterName;
    const filterString = `${field}=${filterName}.${value}`;
    const filterChannel = hasFilter ? ':' + filterString : '';
    const filter = hasFilter ? filterString : undefined;

    // create the callback function
    return (callback: (snap: SupaSnap<T>) => void) => {
      const initialize = () => {
        let select = supabase.from(table).select('*');
        select = hasFilter
          ? select[filterName](field, value as unknown as unknown[])
          : select;

        // match subscription input, with limit
        select.limit(limit).then(({ data, error }) => {
          if (data) items.push(...data);
          callback({
            data: data ? (single ? data[0] : data) : [],
            payload: {
              schema,
              table,
              errors: error,
            } as unknown as Payload,
          });
        });
      };

      const realtimeEvents = (payload: Payload) => {
        switch (payload.eventType) {
          case 'INSERT':
            items.unshift(payload.new);
            break;
          case 'UPDATE':
            let objIndex1 = items.findIndex(
              (obj) => obj[idField] === payload.new[idField]
            );
            if (objIndex1 !== -1) items[objIndex1] = payload.new;
            break;
          case 'DELETE':
            let objIndex2 = items.findIndex(
              (obj) => obj[idField] === payload.old[idField]
            );
            if (objIndex2 !== -1) items.splice(objIndex2, 1);
            break;
        }

        console.log('items', items);
      };

      const channel = supabase
        .channel(`${schema}:${table}${filterChannel}`)
        .on(
          'postgres_changes',
          { event: '*', schema, table, filter },
          (payload) => {
            realtimeEvents(payload);
            // return ALL data with payload
            return callback({ data: single ? items[0] : items, payload });
          }
        )
        .subscribe((status: any) => {
          if (status === 'CHANNEL_ERROR') {
            supabase.removeChannel(channel).then(() => {
              isSubscribed = false;
            });
          }

          if (status === 'SUBSCRIBED') {
            isSubscribed = true;
            initialize();
          }
        });

      return () => {
        isSubscribed = false;
        supabase.removeChannel(channel);
      };
    };
  };

  return {
    from: (table: string) => {
      return {
        subscribe: (callback: (snap: SupaSnap<T>) => void) => {
          document.addEventListener('visibilitychange', () => {
            if (document.visibilityState === 'visible' && !isSubscribed) {
              _subscribe({ table })(callback);
            }
          });

          return _subscribe({ table })(callback);
        },
      };
    },
  };
};