Anyone faced the issue with the Upload queue not performing correctly ?

[abhishekm18-arbadev]

Hi , i'm looking to migrate from MongoDB Atlas Device Sync(realm) to Powersync . I have followed the migration guide which is provided by the official blog and website of powersync itself. I'm facing a strange issue in my app which is a realtime , offline first chat app.

In this , when i send a message i'm performing 3 tasks from frontend associated to it for chat .

1. Write the message payload to the local DB(sqlite) via powersync.execute() in the message table.
2. Update the last message field of the chat table of local DB with the id of the above last message .
3. Update the lastMessageReadTime field on the chat table of local DB with current time when seen for particular user.

I can see that the above transactions seem to perform good mostly when i'm doings chats back and forth at normal pace , like how a normal paced chat is done.
But this gets broken and messages go missing at receiver sides in cases of a group chats or even 1v1 chats when i look to exchange messages rapidly (like keep spamming).

It seems like the uploadData() in the backendConnector is not being called (database.getnextCrudTransaction doesn't return any record) even when the sql update / write transactions complete successfully in the local DB .

Please note that i have checked the backend api's and they are not creating an issue here.

Did anyone else face similar kind of issue ? Or if anyone can help me debug or understand the root cause of messages going missing. Your guidance and help is much appreciated.

[michaelbarnes]

Hey @abhishekm18-arbadev Can you elaborate on the specific of your app setup, please?

• PowerSync Client SDK + version numbers of the SDK and it's peer dependencies.
• Which platform are you testing on.
• Can you share some code snippets on how you're initializing the PowerSyncDatabase and how you're executing the writes into the local SQLite database?

[abhishekm18-arbadev]

Hey @michaelbarnes , please find the details below

PowerSync Client SDK: "@powersync/react-native": "^1.20.1",
React Native: "0.73.6",
React: "18.2.0",
SQLite: "@journeyapps/react-native-quick-sqlite": "^2.4.4",
Query tool: "@powersync/tanstack-react-query": "^0.0.23",
Platform: iOS 18.4.1 / Android 14

Below is System.ts code -

import '@azure/core-asynciterator-polyfill';
import React from 'react';
import { PowerSyncDatabase } from '@powersync/react-native';
import { BackendConnector } from './BackendConnector';
import { AppSchema } from '../powersync-models/schema';

export class System {
powersync: PowerSyncDatabase;
backendConnector: BackendConnector;

constructor() {
this.backendConnector = new BackendConnector();
this.powersync = new PowerSyncDatabase({
schema: AppSchema,
database: {
dbFilename: 'MongoDB.db'
}
});
}

async init() {
await this.powersync.init();
try {
await this.powersync.connect(this.backendConnector);
console.log(Is PowerSync connected: ${JSON.stringify(this.powersync.connected)});
} catch (error) {
console.error("Error connecting to PowerSync:", JSON.stringify(error) );
}

await this.powersync.waitForFirstSync();
console.log("Finished first initial sync.")

}
}

export const system = new System();

export const SystemContext = React.createContext(system);
export const useSystem = () => React.useContext(SystemContext);

And below is BackendConnector.ts -

import {
AbstractPowerSyncDatabase,
CrudEntry,
PowerSyncBackendConnector,
UpdateType,
} from "@powersync/react-native";
import { AppConfig } from "./AppConfig";
import { ApiClient } from "./ApiClient";
import axios from 'axios';

export class BackendConnector implements PowerSyncBackendConnector {
public apiClient: ApiClient;

constructor() {
    this.apiClient = new ApiClient(AppConfig.backendUrl);
}

async fetchCredentials() {

    try {
        return {
            endpoint: AppConfig.powersyncUrl ?? "",
            token: AppConfig.powerSyncDevelopmentToken ?? "",
        };
    } catch (error) {
        console.error('Error fetching PowerSync token:', error);
        return {
            endpoint: "",
            token: "",
        };
    }
}

async uploadData(database: AbstractPowerSyncDatabase): Promise<void> {
    const transaction = await database.getNextCrudTransaction();
    if (!transaction) {
        return;
    }

    let lastOp: CrudEntry | null = null;

    try {
        for (const op of transaction.crud) {
            lastOp = op;
            const record = { table: op.table, data: { ...op.opData, id: op.id } };
            switch (op.op) {
                case UpdateType.PUT:
                    await this.apiClient.upsert(record);
                    break;
                case UpdateType.PATCH:
                    await this.apiClient.update(record);
                    break;
                case UpdateType.DELETE:
                    await this.apiClient.delete(record);
                    break;
            }
        }
        await transaction.complete();
    } catch (ex: any) {
        console.debug(ex);
        console.error("Data upload error - discarding:", lastOp, ex);
    }
}

}

Below is how i'm performing my write transactions in GroupChat -

await powersync.writeTransaction(async (tx) => {
await tx.execute(
INSERT INTO messages (id, senderId, chatId, content, timestamp, messageCategory, messageType, imageUrl, videoUrl, isDeleted, replyTo) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?),
[
mongoId,
user._id.toString(),
chatId1,
messageType == 'text' ? inputMessage : caption,
timestamp,
'message',
messageType,
messageType === 'image' ? imageUri : '',
messageType === 'video' ? imageUri : '',
false,
replyToMessage ? replyToMessage._id.toString() : null
]
);

await tx.execute(
UPDATE chats SET lastMessage = ?, latestMessageTimestamp = ? WHERE id = ?,
[mongoId, new Date().toISOString(), chatId1]
);

// Get current read times
const chatData = await tx.get(
SELECT lastMessageReadTimes FROM chats WHERE id = ?,
[chatId1]
);

  if (!chatData) {
    throw new Error("Chat not found");
  }

  // Parse existing read times or initialize empty array
  let lastMessageReadTimes = [];
  if (chatData.lastMessageReadTimes) {
    try {
      lastMessageReadTimes = JSON.parse(chatData.lastMessageReadTimes);
    } catch(e) {
      console.error("JSON parse error", e);
    }
  }

  // Update read time for current user
  const userIndex = lastMessageReadTimes.findIndex(rt => rt.userId === userId);
  if (userIndex > -1) {
    lastMessageReadTimes[userIndex].lastMessageReadTime = timestamp;
  } else {
    lastMessageReadTimes.push({
      userId: userId,
      lastMessageReadTime: timestamp
    });
  }

  // Update chat with new read times
  await tx.execute(
    UPDATE chats SET lastMessageReadTimes = ? WHERE id = ?,
    [JSON.stringify(lastMessageReadTimes), chatId1]
  );

})

[abhishekm18-arbadev]

@michaelbarnes any update ? or any suggestions ?