Feature/supabase schemas

src/plugins/replication-supabase/index.ts

@@ -58,6 +58,7 @@ export function replicateSupabase<RxDocType>(
     // set defaults
     options.waitForLeadership = typeof options.waitForLeadership === 'undefined' ? true : options.waitForLeadership;
     options.live = typeof options.live === 'undefined' ? true : options.live;
+    const schemaName = typeof options.schemaName === 'undefined' ? 'public' : options.schemaName;
     const modifiedField = options.modifiedField ? options.modifiedField : DEFAULT_MODIFIED_FIELD;
     const deletedField = options.deletedField ? options.deletedField : DEFAULT_DELETED_FIELD;
 
@@ -86,7 +87,7 @@ export function replicateSupabase<RxDocType>(
         return doc;
     }
     async function fetchById(id: string): Promise<WithDeleted<RxDocType>> {
-        const { data, error } = await options.client
+        const { data, error } = await options.client.schema(schemaName)
             .from(options.tableName)
             .select()
             .eq(primaryPath, id)
@@ -103,7 +104,7 @@ export function replicateSupabase<RxDocType>(
                 lastPulledCheckpoint: SupabaseCheckpoint | undefined,
                 batchSize: number
             ) {
-                let query = options.client
+                let query = options.client.schema(schemaName)
                     .from(options.tableName)
                     .select('*');
 
@@ -169,7 +170,7 @@ export function replicateSupabase<RxDocType>(
         ) {
             async function insertOrReturnConflict(doc: WithDeleted<RxDocType>): Promise<WithDeleted<RxDocType> | undefined> {
                 const id = (doc as any)[primaryPath];
-                const { error } = await options.client.from(options.tableName).insert(doc)
+                const { error } = await options.client.schema(schemaName).from(options.tableName).insert(doc)
                 if (!error) {
                     return;
                 } else if (error.code == POSTGRES_INSERT_CONFLICT_CODE) {
@@ -197,7 +198,7 @@ export function replicateSupabase<RxDocType>(
                 // modified field will be set server-side
                 delete toRow[modifiedField];
 
-                let query = options.client
+                let query = options.client.schema(schemaName)
                     .from(options.tableName)
                     .update(toRow);
 
@@ -262,7 +263,7 @@ export function replicateSupabase<RxDocType>(
                 .channel('realtime:' + options.tableName)
                 .on(
                     'postgres_changes',
-                    { event: '*', schema: 'public', table: options.tableName },
+                    { event: '*', schema: schemaName, table: options.tableName },
                     (payload) => {
                         /**
                          * We assume soft-deletes in supabase

src/plugins/replication-supabase/types.ts

@@ -25,6 +25,7 @@ export type SyncOptionsSupabase<RxDocType> = Omit<
 > & {
     client: SupabaseClient;
     tableName: string;
+    schemaName?: string;
 
     /**
      * Modified field, default "_modified"