Промежуточный этап синхронизации

app/components/ChatHeader/ChatHeader.tsx

@@ -31,7 +31,7 @@ export function ChatHeader() {
     const theme = useMantineTheme(); 
     const [blocked, blockUser, unblockUser] = useBlacklist(dialog);
     const [opponent, ___, forceUpdateUserInformation] = useUserInformation(dialog);
-    const protocolState = useProtocolState();
+    const [protocolState] = useProtocolState();
     const [userTypeing, setUserTypeing] = useState(false);
     const timeoutRef = useRef<NodeJS.Timeout>(undefined);
     const avatars = useAvatars(dialog);

app/components/DialogHeaderText/DialogHeaderText.tsx

@@ -0,0 +1,39 @@
+import { useRosettaColors } from "@/app/hooks/useRosettaColors";
+import { ProtocolState } from "@/app/providers/ProtocolProvider/ProtocolProvider";
+import { useProtocolState } from "@/app/providers/ProtocolProvider/useProtocolState";
+import { Flex, Loader, Text } from "@mantine/core";
+
+export function DialogHeaderText() {
+    const [protocolState] = useProtocolState();
+    const colors = useRosettaColors();
+
+    const headerType = () => {
+        switch(protocolState){
+            case ProtocolState.SYNCHRONIZATION:
+                return (<>
+                    <Loader size={12} color={colors.chevrons.active}></Loader>
+                    <Text fw={500} style={{
+                        userSelect: 'none'
+                    }} size={'sm'}>Updating...</Text>
+                </>);
+            case ProtocolState.CONNECTED:
+                return (<>
+                    <Text fw={500} style={{
+                        userSelect: 'none'
+                    }} size={'sm'}>Chats</Text>
+                </>);
+            default:
+                return (<>
+                    <Text fw={500} style={{
+                        userSelect: 'none'
+                    }} size={'sm'}>Chats</Text>
+                </>);
+        }
+    }
+
+    return (
+        <Flex direction={'row'} align={'center'} gap={'xs'}>
+            {headerType()}
+        </Flex>
+    )
+}

app/components/DialogsPanelHeader/DialogsPanelHeader.tsx

@@ -6,6 +6,7 @@ import { useLogout } from "@/app/providers/AccountProvider/useLogout";
 import { useHotkeys } from "@mantine/hooks";
 import { useNavigate } from "react-router-dom";
 import { usePublicKey } from "@/app/providers/AccountProvider/usePublicKey";
+import { DialogHeaderText } from "../DialogHeaderText/DialogHeaderText";
 
 export function DialogsPanelHeader() {
     const colors = useRosettaColors();
@@ -66,9 +67,7 @@ export function DialogsPanelHeader() {
                         </Menu.Item>
                     </Menu.Dropdown>
                 </Menu>
-                <Text fw={500} style={{
+                <DialogHeaderText></DialogHeaderText>
-                    userSelect: 'none'
-                }} size={'sm'}>Chats</Text>
                 <Menu withArrow width={150} shadow="md">
                     <Menu.Target>
                         <IconEdit style={{

app/components/GroupHeader/GroupHeader.tsx

@@ -25,7 +25,7 @@ export function GroupHeader() {
     const {deleteMessages, dialog} = useDialog();
     const theme = useMantineTheme(); 
     const {groupInfo} = useGroupInformation(dialog);
-    const protocolState = useProtocolState();
+    const [protocolState] = useProtocolState();
     const [usersTypeing, setUsersTypeing] = useState<{
         timeout: NodeJS.Timeout | null,
         fromPublicKey: string

app/components/Topbar/Topbar.tsx

@@ -8,7 +8,7 @@ import { MacFrameButtons } from "../MacFrameButtons/MacFrameButtons";
 
 export function Topbar() {
     const colors = useRosettaColors();
-    const protocolState = useProtocolState(); 
+    const [protocolState] = useProtocolState(); 
 
 
     return (
@@ -16,14 +16,14 @@ export function Topbar() {
             {window.platform == 'win32' && <WindowsFrameButtons></WindowsFrameButtons>}
             {window.platform == 'darwin' && <MacFrameButtons></MacFrameButtons>}
             {window.platform == 'linux' && <WindowsFrameButtons></WindowsFrameButtons>}
-            {(protocolState == ProtocolState.CONNECTED || !window.location.hash.includes("main")) && 
+            {(protocolState == ProtocolState.CONNECTED || protocolState == ProtocolState.SYNCHRONIZATION || !window.location.hash.includes("main")) && 
                 <Flex align={'center'} justify={'center'}>
                     <Text fw={'bolder'} fz={13} c={'gray'}>
                         Rosetta Messenger
                     </Text>
                 </Flex>
             }
-            {(protocolState != ProtocolState.CONNECTED && protocolState != ProtocolState.DEVICE_VERIFICATION_REQUIRED && window.location.hash.includes("main")) && 
+            {(protocolState != ProtocolState.CONNECTED && protocolState != ProtocolState.SYNCHRONIZATION && protocolState != ProtocolState.DEVICE_VERIFICATION_REQUIRED && window.location.hash.includes("main")) && 
                 <Flex align={'center'} gap={5} justify={'center'}>
                     <Loader size={12} color={colors.chevrons.active}></Loader>
                     <Text fw={'bolder'} fz={13} c={'gray'}>

app/providers/DialogProvider/dialogQueue.ts

@@ -0,0 +1,12 @@
+let tail: Promise<void> = Promise.resolve();
+
+export const runTaskInQueue = (fn: () => Promise<void>) => {
+    tail = tail.then(fn).catch((e) => {
+        console.error("Dialog queue error", e);
+    });
+};
+
+/**
+ * Ждет, пока все пакеты попадающие в очередь не будут обработаны
+ */
+export const whenFinish = () => tail;

app/providers/DialogProvider/useDialog.ts

@@ -44,7 +44,7 @@ export function useDialog() : {
     const privateKey = usePrivateKeyHash();
     const privatePlain = usePrivatePlain();
     const {writeFile} = useFileStorage();
-    const protocolState = useProtocolState();
+    const [protocolState] = useProtocolState();
     const {hasGroup, getGroupKey} = useGroups();
     const {warn} = useConsoleLogger('useDialog');
 

app/providers/DialogProvider/useDialogFiber.ts

@@ -24,6 +24,7 @@ import { useGroups } from "./useGroups";
 import { useDialogState } from "../DialogStateProvider.tsx/useDialogState";
 import { useUserInformation } from "../InformationProvider/useUserInformation";
 import { useMentions } from "../DialogStateProvider.tsx/useMentions";
+import { runTaskInQueue } from "./dialogQueue";
 
 /**
  * При вызове будет запущен "фоновый" обработчик
@@ -64,6 +65,7 @@ export function useDialogFiber() {
      * Метод нужен для синхронизации своих сообщений
      */
     usePacket(0x06, async (packet: PacketMessage) => {
+        runTaskInQueue(async () => {
             const fromPublicKey = packet.getFromPublicKey();
             const toPublicKey = packet.getToPublicKey();
             const aesChachaKey = packet.getAesChachaKey();
@@ -148,12 +150,14 @@ export function useDialogFiber() {
             if (currentDialogPublicKeyView !== toPublicKey && dialogCache.length > 0) {
                 addOrUpdateDialogCache(toPublicKey, [...dialogCache, newMessage].slice(-MESSAGE_MAX_LOADED));
             }
+        });
     }, [privatePlain, currentDialogPublicKeyView]);
 
     /**
      * Обработчик сообщений для группы
      */
     usePacket(0x06, async (packet: PacketMessage) => {
+        runTaskInQueue(async () => {
             const fromPublicKey = packet.getFromPublicKey();
             const toPublicKey = packet.getToPublicKey();
             const content = packet.getContent();
@@ -298,11 +302,13 @@ export function useDialogFiber() {
             if (currentDialogPublicKeyView !== toPublicKey && dialogCache.length > 0) {
                 addOrUpdateDialogCache(toPublicKey, [...dialogCache, newMessage].slice(-MESSAGE_MAX_LOADED));
             }
+        });
     }, [blocked, muted, updateDialog, focused, currentDialogPublicKeyView, viewState, idle]);
     /**
      * Обработчик личных сообщений
      */
     usePacket(0x06, async (packet: PacketMessage) => {
+        runTaskInQueue(async () => {
             const fromPublicKey = packet.getFromPublicKey();
             if (fromPublicKey == publicKey) {
                 /**
@@ -423,12 +429,14 @@ export function useDialogFiber() {
             if (currentDialogPublicKeyView !== fromPublicKey && dialogCache.length > 0) {
                 addOrUpdateDialogCache(fromPublicKey, [...dialogCache, newMessage].slice(-MESSAGE_MAX_LOADED));
             }
+        });
     }, [blocked, muted, updateDialog, focused, currentDialogPublicKeyView, viewState, idle]);
 
     /**
      * Обработчик синхронизации прочтения личных сообщений
      */
     usePacket(0x07, async (packet: PacketRead) => {
+        runTaskInQueue(async () => {
             if (hasGroup(packet.getToPublicKey())) {
                 /**
                  * Если это относится к группам, то игнорируем здесь,
@@ -462,12 +470,14 @@ export function useDialogFiber() {
                 }
                 return message;
             }));
+        });
     }, [updateDialog, publicKey]);
 
     /**
      * Обработчик прочтения личных сообщений
      */
     usePacket(0x07, async (packet: PacketRead) => {
+        runTaskInQueue(async () => {
             if (hasGroup(packet.getToPublicKey())) {
                 /**
                  * Если это относится к группам, то игнорируем здесь,
@@ -486,6 +496,7 @@ export function useDialogFiber() {
             }
             console.info("PACKED_READ_IM");
             await runQuery(`UPDATE messages SET read = 1 WHERE from_public_key = ? AND to_public_key = ? AND account = ?`, [toPublicKey, fromPublicKey, publicKey]);
+            console.info("read im with params ", [fromPublicKey, toPublicKey, publicKey]);
             updateDialog(fromPublicKey);
             log("Read packet received from " + fromPublicKey + " for " + toPublicKey);
             addOrUpdateDialogCache(fromPublicKey, getDialogCache(fromPublicKey).map((message) => {
@@ -499,11 +510,13 @@ export function useDialogFiber() {
                 }
                 return message;
             }));
+        });
     }, [updateDialog, publicKey]);
     /**
      * Обработчик прочтения групповых сообщений
      */
     usePacket(0x07, async (packet: PacketRead) => {
+        runTaskInQueue(async () => {
             if (!hasGroup(packet.getToPublicKey())) {
                 /**
                  * Если это не относится к группам, то игнорируем здесь,
@@ -526,11 +539,13 @@ export function useDialogFiber() {
                 }
                 return message;
             }));
+        });
     }, [updateDialog]);
     /**
      * Обработчик доставки сообщений
      */
     usePacket(0x08, async (packet: PacketDelivery) => {
+        runTaskInQueue(async () => {
             const messageId = packet.getMessageId();
             await runQuery(`UPDATE messages SET delivered = ?, timestamp = ? WHERE message_id = ? AND account = ?`, [DeliveredMessageState.DELIVERED, Date.now(), messageId, publicKey]);
             updateDialog(packet.getToPublicKey());
@@ -545,5 +560,6 @@ export function useDialogFiber() {
                 }
                 return message;
             }));
+        });
     }, [updateDialog]);
 }

app/providers/DialogProvider/useSynchronize.ts

@@ -0,0 +1,57 @@
+import { useEffect } from "react";
+import { useProtocolState } from "../ProtocolProvider/useProtocolState";
+import { ProtocolState } from "../ProtocolProvider/ProtocolProvider";
+import { useDatabase } from "../DatabaseProvider/useDatabase";
+import { usePublicKey } from "../AccountProvider/usePublicKey";
+import { PacketSync, SyncStatus } from "../ProtocolProvider/protocol/packets/packet.sync";
+import { useSender } from "../ProtocolProvider/useSender";
+import { usePacket } from "../ProtocolProvider/usePacket";
+import { whenFinish } from "./dialogQueue";
+
+/**
+ * Хук отвечает за синхронизацию сообщений, запрос синхронизации
+ * при подключении
+ */
+export function useSynchronize() {
+    const [protocolState, setProtocolState] = useProtocolState();
+    const {getQuery} = useDatabase();
+    const publicKey = usePublicKey();
+    const send = useSender();
+
+    useEffect(() => {
+        if(protocolState == ProtocolState.CONNECTED){
+            trySync();
+            setProtocolState(ProtocolState.SYNCHRONIZATION);
+        }
+    }, [protocolState]);
+
+    const trySync = async () => {
+        const lastMessage = await getQuery("SELECT timestamp FROM messages WHERE account = ? ORDER BY timestamp DESC LIMIT 1", [publicKey]);
+        if(!lastMessage){
+            sendSynchronize(0);
+            return;
+        }
+        sendSynchronize(lastMessage.timestamp);
+    }
+
+    const sendSynchronize = (timestamp: number) => {
+        const packet = new PacketSync();
+        packet.setStatus(0);
+        packet.setTimestamp(timestamp);
+        send(packet);
+    }
+
+    usePacket(25, async (packet: PacketSync) => {
+        const status = packet.getStatus();
+        if(status == SyncStatus.BATCH_END){
+            await whenFinish();
+            trySync();
+        }
+        if(status == SyncStatus.NOT_NEEDED){
+            /**
+             * Синхронизация не нужна, все данные актуальны
+             */
+            setProtocolState(ProtocolState.CONNECTED);
+        }
+    });
+}

app/providers/ProtocolProvider/ProtocolProvider.tsx

@@ -12,10 +12,13 @@ export enum ProtocolState {
     HANDSHAKE_EXCHANGE,
     DISCONNECTED,
     RECONNECTING,
-    DEVICE_VERIFICATION_REQUIRED
+    DEVICE_VERIFICATION_REQUIRED,
+    SYNCHRONIZATION
 }
 
-export const ProtocolContext = createContext<[Protocol|null, ProtocolState]>([null, ProtocolState.DISCONNECTED]);
+export type ProtocolContextType = [Protocol|null, ProtocolState, (state: ProtocolState) => void];
+
+export const ProtocolContext = createContext<ProtocolContextType>([null, ProtocolState.DISCONNECTED, () => {}]);
 
 interface ProtocolProviderProps {
     children: React.ReactNode;
@@ -91,7 +94,7 @@ export function ProtocolProvider(props : ProtocolProviderProps) {
     }, [publicKey, privateKey, systemInfo.id]);
 
     return (
-        <ProtocolContext.Provider value={[protocol, connect]}>
+        <ProtocolContext.Provider value={[protocol, connect, setConnect]}>
             {props.children}
         </ProtocolContext.Provider>
     );

app/providers/ProtocolProvider/protocol/packets/packet.sync.ts

@@ -0,0 +1,48 @@
+import Packet from "../packet";
+import Stream from "../stream";
+
+export enum SyncStatus {
+    NOT_NEEDED,
+    BATCH_START,
+    BATCH_END
+}
+
+export class PacketSync extends Packet {
+
+    private status : SyncStatus = SyncStatus.NOT_NEEDED;
+    private timestamp : number = 0;
+
+    public getPacketId(): number {
+        return 25; //0x19
+    }
+
+    public _receive(stream: Stream): void {
+        this.status = stream.readInt8() as SyncStatus;
+        this.timestamp = stream.readInt64();
+    }
+
+    public _send(): Promise<Stream> | Stream {
+        let stream = new Stream();
+        stream.writeInt16(this.getPacketId());
+        stream.writeInt8(this.status);
+        stream.writeInt64(this.timestamp);
+        return stream;
+    }
+
+    public getStatus() : SyncStatus {
+        return this.status;
+    }
+
+    public setStatus(status: SyncStatus) {
+        this.status = status;
+    }
+
+    public getTimestamp() : number {
+        return this.timestamp;
+    }
+
+    public setTimestamp(timestamp: number) {
+        this.timestamp = timestamp;
+    }
+    
+}

app/providers/ProtocolProvider/protocol/protocol.ts

@@ -24,6 +24,7 @@ import { PacketGroupBan } from "./packets/packet.group.ban";
 import { PacketDeviceNew } from "./packets/packet.device.new";
 import { PacketDeviceList } from "./packets/packet.device.list";
 import { PacketDeviceResolve } from "./packets/packet.device.resolve";
+import { PacketSync } from "./packets/packet.sync";
 
 export default class Protocol extends EventEmitter {
     private serverAddress: string;
@@ -123,6 +124,7 @@ export default class Protocol extends EventEmitter {
         this._supportedPackets.set(0x16, new PacketGroupBan());
         this._supportedPackets.set(0x17, new PacketDeviceList());
         this._supportedPackets.set(0x18, new PacketDeviceResolve());
+        this._supportedPackets.set(25, new PacketSync());
     }
 
     private _findWaiters(packetId: number): ((packet: Packet) => void)[] {

app/providers/ProtocolProvider/useProtocolState.ts

@@ -1,12 +1,12 @@
 import { useContext } from "react";
-import { ProtocolContext } from "./ProtocolProvider";
+import { ProtocolContext, ProtocolContextType, ProtocolState } from "./ProtocolProvider";
 
 export const useProtocolState = () => {
-    const [context, connect] = useContext(ProtocolContext);
+    const context : ProtocolContextType = useContext(ProtocolContext);
 
     if(!context){
         throw new Error("useProtocol must be used within a ProtocolProvider");
     }
 
-    return connect;
+    return [context[1], context[2]] as [ProtocolState, (state: ProtocolState) => void];
 };

app/views/DeviceConfirm/DeviceConfirm.tsx

@@ -12,7 +12,7 @@ import { usePacket } from "@/app/providers/ProtocolProvider/usePacket";
 import { PacketDeviceResolve, Solution } from "@/app/providers/ProtocolProvider/protocol/packets/packet.device.resolve";
 
 export function DeviceConfirm() {
-    const protocolState = useProtocolState();
+    const [protocolState] = useProtocolState();
     const navigate = useNavigate();
     const logout = useLogout();
 

app/views/Main/Main.tsx

@@ -30,6 +30,7 @@ import { useLogout } from "@/app/providers/AccountProvider/useLogout";
 import { useUpdateMessage } from "@/app/hooks/useUpdateMessage";
 import { useDeviceMessage } from "@/app/hooks/useDeviceMessage";
 import { UpdateProvider } from "@/app/providers/UpdateProvider/UpdateProvider";
+import { useSynchronize } from "@/app/providers/DialogProvider/useSynchronize";
 
 export function Main() {
     const { mainColor, borderColor } = useRosettaColors();
@@ -54,6 +55,11 @@ export function Main() {
      */
     useDeviceMessage();
 
+    /**
+     * Синхронизируем сообщения при подключении
+     */
+    useSynchronize();
+
     const { setSize, setResizeble } = useWindow();
 
     /**