Downstream updates not working using subscribe

[jasondavis87]

Hey everyone, I'll start off by saying this is probably more a react thing than a watermelon thing, but i'm not sure which way to go. I've used https://github.com/bndkt/sharemystack and the watermelon docs as a reference and came up with a custom hook:

THE CUSTOM HOOK

"use client";

import { useEffect, useState } from "react";
import { useDatabase } from "@nozbe/watermelondb/hooks";

import type { Post } from "@repo/database";
import { TableName } from "@repo/database";

import { useSync } from "./use-sync";

export function usePost(id: string) {
  const database = useDatabase();
  const sync = useSync();
  const [post, setPost] = useState<Post | null>(null);

  useEffect(() => {
    const sub = database
      .get<Post>(TableName.POSTS)
      .findAndObserve(id)
      .subscribe((data) => {
        console.log("usePost", data);
        setPost(data ?? null);
      });

    return () => sub.unsubscribe();
  }, [database, sync]);


  return { post };
}

USING THE HOOK

"use client";

import { useEffect } from "react";
import { usePost } from "../hooks/usePost";

interface ClientPostPageProps {
  id: string;
}

const ClientPostPage = (props: ClientPostPageProps) => {
  const { id } = props;
  const { post } = usePost(id);

  useEffect(() => {
    console.log("useEffect", post);
  }, [post])

  if (!post) {
    return null;
  }

  return (
    <main>
      Testing
    </main>
  );
}

• I'm using Typescript
• I'm using NextJS 14 App router
• I'm using Turborepo to build the schema/models/decorators, minus the connector
• I'm pretty sure the model/schema is setup correctly since im able to pull data on first load
• I DO get the usePost, data console log when data updates (I've got this syncing w/ supabase realtime)
• PROBLEM: I DON'T get any other downstream updates, i.e. i never get the useEffect console log. I only get that first initial load.

Any help is appreciated, thanks in advance :)

[UrsDeSwardt]

Hey! I'm having trouble getting Watermelon to work with Nextjs. Do you have your DatabaseProvider setup correctly?

[heliocosta-dev]

Do you mind sharing the content of useSync?

[jasondavis87]

Hey! I'm having trouble getting Watermelon to work with Nextjs. Do you have your DatabaseProvider setup correctly?

I've got some caveats because im building my database in a turbo repo package and then importing into both a mobile (which i havent built yet) and nextjs app. I think the part you're looking for mostly is the layout.tsx. It's what took me the longest to figure out.

@/lib/database/index.ts

"use client";

import { createClient } from "@/lib/supabase/client";
import { Database } from "@nozbe/watermelondb";
import LokiJSAdapter from "@nozbe/watermelondb/adapters/lokijs";
import { synchronize } from "@nozbe/watermelondb/sync";
import { setGenerator } from "@nozbe/watermelondb/utils/common/randomId";
import { v4 } from "uuid";

import { migrations, models, schema } from "@repo/database";

// First, create the adapter to the underlying database:
const adapter = new LokiJSAdapter({
  schema,
  // (You might want to comment out migrations for development purposes -- see Migrations documentation)
  //migrations,
  useWebWorker: false,
  useIncrementalIndexedDB: true,
  // dbName: 'myapp', // optional db name

  // --- Optional, but recommended event handlers:

  onQuotaExceededError: (error) => {
    // Browser ran out of disk space -- offer the user to reload the app or log out
    console.error("[watermelondb.QuotaExceeded]", error);
  },
  onSetUpError: (error) => {
    // Database failed to load -- offer the user to reload the app or log out
    console.error("[watermelondb.SetUpError]", error);
  },
  extraIncrementalIDBOptions: {
    onDidOverwrite: () => {
      // Called when this adapter is forced to overwrite contents of IndexedDB.
      // This happens if there's another open tab of the same app that's making changes.
      // Try to synchronize the app now, and if user is offline, alert them that if they close this
      // tab, some data may be lost
      console.log("Database was overwritten in another tab");
    },
    onversionchange: () => {
      // database was deleted in another browser tab (user logged out), so we must make sure we delete
      // it in this tab as well - usually best to just refresh the page
      // if (checkIfUserIsLoggedIn()) {
      window.location.reload();
      // }
      console.log("Database was deleted in another tab");
    },
  },
});

export const database = new Database({
  adapter,
  modelClasses: [
    // This is where you list all the models in your app
    // that you want to use in the database
    ...models,
  ],
});

setGenerator(() => v4());

export const sync = async () => {
  const supabase = createClient();

  await synchronize({
    database,
    sendCreatedAsUpdated: true,
    pullChanges: async (props) => {
      const { lastPulledAt, schemaVersion, migration } = props;

      const { data, error } = await supabase.rpc("pull", {
        last_pulled_at: lastPulledAt,
        schema_version: schemaVersion,
        migration,
      });

      if (error) {
        throw new Error(`Pull Changes: ${error}`);
      }

      return data;
    },
    pushChanges: async (props) => {
      const { changes, lastPulledAt } = props;

      // use sorted changes to take out any tables that shouldnt be synced
      //const sortedChanges = { programs: changes.programs, ...changes };

      // NOTE: supabase-js has error set as :any
      // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
      const { error } = await supabase.rpc("push", {
        changes,
        last_pulled_at: lastPulledAt,
      });

      if (error) {
        throw new Error(`Push Changes: ${error}}`);
      }
    },
  });
};

@/components/providers/database-provider.tsx

"use client";

import { database } from "@/lib/database";
import { DatabaseProvider as WatermelonDatabaseProvider } from "@nozbe/watermelondb/DatabaseProvider";

interface Props {
  children: React.ReactNode;
}

const DatabaseProvider = (props: Props) => {
  const { children } = props;

  return (
    <WatermelonDatabaseProvider database={database}>
      {children}
    </WatermelonDatabaseProvider>
  );
};

export default DatabaseProvider;

layout.tsx

import React from "react";
import dynamic from "next/dynamic";
import { redirect } from "next/navigation";
import { SyncProvider } from "@/components/providers/sync-provider";
import { createClient } from "@/lib/supabase/server";

const DatabaseProvider = dynamic(
  () => import("../../components/providers/database-provider"),
  { ssr: false },
);

interface MainAppLayoutProps {
  children: React.ReactNode;
}

const MainAppLayout = async (props: Readonly<MainAppLayoutProps>) => {
  const { children } = props;

  // make sure user is signed in
  const supabase = createClient();
  const { data, error } = await supabase.auth.getUser();

  if (error ?? !data?.user) {
    redirect("/signin");
  }

  return (
    <DatabaseProvider>
      <SyncProvider>
        <div className="container p-0">{children}</div>
      </SyncProvider>
    </DatabaseProvider>
  );
};

export default MainAppLayout;

Hope this helps :)

[jasondavis87]

Do you mind sharing the content of useSync?

The goal when i made this was that in my app i can call this however many times i need and it'd only allow 1 sync call to be queued, so if i call sync 20 times whilst it's already syncing, it'll only execute it 1 more time after it's finished. (honestly there's probably a better way to get this done, but this works :)

sync-provider.tsx

"use client";

import type { RealtimeChannel } from "@supabase/supabase-js";
import React, { createContext, useCallback, useEffect, useRef } from "react";
import { sync as watermelon_sync } from "@/lib/database";
import { createClient } from "@/lib/supabase/client";

// create supabase client
const supabase = createClient();

// Create the context with an option function type
export const SyncContext = createContext<(() => Promise<void>) | null>(null);

interface SyncProviderProps {
  children: React.ReactNode;
}

export const SyncProvider = (props: SyncProviderProps) => {
  const { children } = props;

  const isSyncingRef = useRef(false);
  const syncQueueRef = useRef(false);

  const sync = useCallback(async () => {
    if (isSyncingRef.current) {
      syncQueueRef.current = true; // Mark to run sync again after current one finishes
      return;
    }

    isSyncingRef.current = true;
    syncQueueRef.current = false; // Reset the queue

    await watermelon_sync();

    isSyncingRef.current = false;

    // If a sync was queued during the last sync, run it now
    if (syncQueueRef.current) {
      syncQueueRef.current = false; // Reset the queue before starting new sync
      await sync(); // Start new sync if queued
    }
  }, []);

  useEffect(() => {
    console.log("Realtime Initializing");
    let sub: RealtimeChannel | null = null;
    const f = async () => {
      // subscribe to updates
      const { data, error } = await supabase.auth.getUser();
      if (error) {
        console.error("Error fetching user:", error);
        return;
      }
      if (!data?.user) {
        console.error("No user data available.");
        return;
      }

      sub = supabase
        .channel(`realtime:${data.user.id}`)
        .on(
          "postgres_changes",
          {
            event: "*",
            schema: "data",
          },
          () => {
            sync().catch(console.error);
          },
        )
        .subscribe();
    };
    void f();

    return () => {
      if (sub) {
        supabase.removeChannel(sub).catch(console.error);
      }
    };
  }, [sync]);

  return <SyncContext.Provider value={sync}>{children}</SyncContext.Provider>;
};

useSync.tsx

"use client";

import { useContext, useEffect } from "react";
import { SyncContext } from "@/components/providers/sync-provider";

export const useSync = (): (() => Promise<void>) => {
  const sync = useContext(SyncContext);
  if (sync === null) {
    throw new Error("useSync must be used within a SyncProvider");
  }

  useEffect(() => {
    // call sync on mount
    void sync();
  }, [sync]);

  return sync;
};

[jasondavis87]

Back to the original concern. I think i've got a workaround but i'd like an actual fix so if anyone knows one, help me out :). Otherwise here's my quick fix:

import { useReducer } from "react";
const forceUpdate = useReducer(() => ({}), {})[1] as () => void;

Found on stack overflow to force an update. I put this in every subscription function, i.e.:

useEffect(() => {
    const sub = database
      .get<Post>(TableName.POSTS)
      .findAndObserve(id)
      .subscribe((data) => {
        forceUpdate();
        setPost(data ?? null);
      });

    return () => sub.unsubscribe();
  }, [database, sync]);

[UrsDeSwardt]

@jasondavis87 I did something similar with a custom hook that's working for me:

import { Observable } from "@nozbe/watermelondb/utils/rx";
import { useEffect, useState } from "react";
import { Post, Comment } from "@/db/models";
import { useDatabase } from "@nozbe/watermelondb/react";
import { Q } from "@nozbe/watermelondb";

const defaultObservable = <T>(): Observable<T[]> =>
  new Observable<T[]>((observer) => {
    observer.next([]);
  });

export const useGetPosts = () => {
  const database = useDatabase();
  const [posts, setPosts] = useState<Observable<Post[]>>(defaultObservable);

  useEffect(() => {
    setPosts(database.get<Post>(Post.table).query().observe());
  }, [database]);

  return posts;
};

[jasondavis87]

Awesome, let me try that and report back.

[heliocosta-dev]

@UrsDeSwardt have you tried observeWithColumns?

[UrsDeSwardt]

@heliocosta-dev I tried it now with no luck

[UrsDeSwardt]

Here's my full example:
https://github.com/UrsDeSwardt/watermelondb-nextjs-example

[KrisLau]

EDIT: My project is a React Native project so I'm not too sure if it'll work in Next but it should be ok?

The goal when i made this was that in my app i can call this however many times i need and it'd only allow 1 sync call to be queued, so if i call sync 20 times whilst it's already syncing, it'll only execute it 1 more time after it's finished. (honestly there's probably a better way to get this done, but this works :)

if you want to debounce the sync here's how I did mine (with RxJs):

database
    .withChangesForTables([
         'table_1',
         'table_2',
         // ...other tables in db
    ])
    .pipe(
      skip(1),
      // ignore records simply becoming `synced`
      filter(changes => !changes.every(change => change.record.syncStatus === 'synced')),
      // debounce to avoid syncing in the middle of related actions
      debounceTime(1000),
    )
    .subscribe(async () => {
      await sync(); // calls the method which does the synchronizeWithServer call
    });

For your main question of observing updates, the doc has a section on observing updates here. Example from a component in my project:

import {Q} from '@nozbe/watermelondb';
import {withDatabase, withObservables} from '@nozbe/watermelondb/react';
import compose from '@shopify/react-compose';

const ExampleComponent = ({user}) => { // user is coming from the observable query as a prop
    // component
}

export default compose(
  withDatabase,
  withObservables([], ({database}) => ({
    user: database
      .get('user')
      .query(Q.where('id', userID))
      .observeWithColumns(['first_name', 'last_name']),
  })),
)(ExampleComponent);

[UrsDeSwardt]

@KrisLau does that actually work for you for observing updates using NextJS?

[KrisLau]

@UrsDeSwardt Sorry I should've specified that I use it in React Native, not in Next! (Edited my original comment to add the context)

The code should still work though I think?

[UrsDeSwardt]

@KrisLau Thanks for the update! I also come from RN where this approach worked perfectly. However, I couldn't get it to work with NextJS. The only way I get it to work is by using the hook I mentioned above.

[isaachinman]

Thought I'd post this here. It's rather silly that Watermelon doesn't ship hooks out of the box. The documentation makes it sound like HOC=>hooks is some massive leap. I suspect instead that the repo hasn't seen much maintenance in recent years. You can achieve a flexible and reusable hook in like 10 lines of code. You don't need to use a database provider or context.

import { useMemo } from 'react'

import { Clause } from '@nozbe/watermelondb/QueryDescription'
import { useObservableState } from 'observable-hooks'

export const useDatabaseRows = <T extends keyof TableConfig>(
  tableName: T,
  _query: Clause[] = [],
  _observableColumns: TableConfig[T]['columnNames'][] = [],
) => {
  const observableColumns = useMemo(() => _observableColumns, [_observableColumns.join(',')])
  const query = useMemo(() => _query, [JSON.stringify(_query)])

  const observable = useMemo(
    () => database
      .get<Model>(tableName)
      .query(query)
      .observeWithColumns(observableColumns),
    [tableName, observableColumns, query],
  )

  return useObservableState(observable, []) as TableConfig[T]['value']
}

@radex Happy to put together a PR for first-class hooks, if you are around to review and advise. I think it'd save a lot of users a lot of headaches.

[heliocosta-dev]

Nice one @isaachinman. What about unsubscribing from these observers? E.g. when one wants to logout & call unsafeResetDatabase?

[isaachinman]

Docs are here.

I gave a better alternative to unsafeResetDatabase in #102 (comment).