Record duplicating while synchronization

[Kam125]

Hey Guys, I'm facing the record duplication error.

Backend: Have integer and sequenced id's
Frontend: Watermelon DB managed id's

In table's I've server_id to represent backend generated id.

Step 01: I've created a record on watermelon db. A record is created with server_id null.

Step 02: Sync function is called, I checked if the records being pushed to server don't have server_id, then get them synced. Server returns me two id's. i) client_id (which is watermelon id) ii) id (backend generated id).

Step 03: Based on client_id, I find record in watermelon db and updated that record server_id with id returned from server.

Step 04: I pulled changes, server returned me two things, created array and updated array, both have same information, and both have id's which server created, only way to differentiate between them is, before creating a new record on watermelon db, I've to check, if I've server_id matching with created array object id on watermelon db, then ignores it. However for updated record I'm not doing any checks, and watermelon db handles that logic.

Step 05: Now after above pull, a new record get's created with same information as I've above, with new watermelon db id and server_id null. Now it synced with server and cycle repeats.

Step 06: On 2nd pull, I get that record only in created array and not in updated array, and new duplication doesn't happens.

That's how things are going up, I tried debugging and logging and it's the only thing I found, If anyone knows what's causing the issues. Please reply to this.

Here's my sync code

`import { synchronize } from "@nozbe/watermelondb/sync";
import { client } from "../utils/api-client";
import AsyncStorage from "@react-native-async-storage/async-storage";
import { database } from "./database";
import { Q } from "@nozbe/watermelondb";

/**

• Transform server response to match WatermelonDB schema
  */
  const transformServerResponse = (serverData: any, tableName: string) => {
  switch (tableName) {
  case "locations":
  return {
  id: serverData.id.toString(),
  name: serverData.name,
  address: serverData.address,
  organization: serverData.organization,
  server_id: serverData.id,
  synced: true,
  created_date: serverData.created_date || new Date().toISOString(),
  item_count: serverData.item_count || 0,
  };

  default:
  return serverData;
  }
  };

export const syncDatabase = async ({ lastPulledAt, orgId }: any) => {
await synchronize({
database,
pullChanges: async () => {
const response = await client(sync/pull/${orgId}/, {
method: "POST",
data: {
lastPulledAt,
schemaVersion: 1,
migrations: null,
},
});

  if (!response) {
    throw new Error("Failed to pull changes");
  }

  console.log("response", JSON.stringify(response));

  const data = await response;
  await AsyncStorage.setItem(
    "lastPulledAt",
    JSON.stringify(data?.timestamp || "")
  );

  // Get existing records with server_ids for each table
  const existingRecords = {
    locations: await database
      .get("locations")
      .query(Q.where("server_id", Q.notEq(null)))
      .fetch(),
  };

  // Modified reduce function with more explicit accumulator initialization
  const serverIdMaps = Object.entries(existingRecords).reduce(
    (acc: any, [table, records]) => {
      const serverIds = records
        .map((record: any) => record._raw.server_id?.toString())
        .filter(Boolean);

      acc[table] = serverIds;

      return acc;
    },
    {} as Record<string, Set<string>> // Explicitly type the initial accumulator
  );

  // Filter and transform changes for "locations"
  const changes = {
    locations: {
      created:
        data.changes.locations?.created
          ?.filter(
            (location: any) =>
              !serverIdMaps.locations.includes(location.id?.toString())
          )
          ?.map((location: any) => transformServerResponse(location, "locations")) ||
        [],
      updated: data.changes.locations?.updated || [],
      deleted: data.changes.locations?.deleted || [],
    },
  };

  return {
    changes,
    timestamp: data.timestamp,
  };
},
pushChanges: async ({ changes }) => {
  console.log("changes---------too push", changes?.locations?.created);

  // Transform local changes for server
  const transformedChanges = {
    locations: {
      created:
        changes.locations?.created
          ?.filter((location) => !location.server_id) // Only include locations that haven't been synced
          ?.map((location) => ({
            client_id: location.id,
            name: location.name,
            address: location.address,
            organization: location.organization,
          })) || [],
      updated:
        changes.locations?.updated
          ?.filter((location) => location.server_id) // Only include locations that have been synced
          ?.map((location) => ({
            id: location.server_id,
            name: location.name,
            address: location.address,
            organization: location.organization,
          })) || [],
      deleted: changes.locations?.deleted || [],
    },
  };

  console.log(
    "transformedChanges---------too push",
    JSON.stringify(transformedChanges?.locations)
  );

  // Only send changes if there are any valid changes
  const hasChanges = Object.values(transformedChanges).some(
    (table) =>
      table.created.length > 0 ||
      table.updated.length > 0 ||
      table.deleted.length > 0
  );

  if (!hasChanges) {
    return;
  }
  const response = await client(`sync/push/`, {
    method: "POST",
    data: { changes: transformedChanges },
  });

  console.log("response from push changes", JSON.stringify(response));

  if (response.message !== "Changes pushed successfully") {
    throw new Error("Failed to push changes");
  }

  // Update local records with server IDs
  try {
    await database.write(async () => {
      // Update Locations and their relationships
      if (response.created?.locations?.length) {
        for (const location of response.created.locations) {
          const locationRecord = await database
            .get("locations")
            .find(location.client_id);
          await locationRecord.update((record: any) => {
            record.server_id = location.id;
            record.synced = true;
            record._status = ""; // Clear status
            record._changed = ""; // Clear changed fields
          });
        }
      }
    });
  } catch (error) {
    console.error("Error updating local records with server IDs:", error);
    throw new Error("Failed to update local records with server IDs");
  }
},

});
};
`