From ad50ca8644e6f1700c5ad3bdf440dd96b2489de7 Mon Sep 17 00:00:00 2001 From: "Evgeny @ SimpleX Chat" <259188159+evgeny-simplex@users.noreply.github.com> Date: Sun, 31 May 2026 06:43:05 +0000 Subject: [PATCH] fixes --- smp-web/src/agent/store-idb.ts | 88 ++++++++++++++++++++-------------- smp-web/src/agent/store.ts | 16 +++---- smp-web/tests/store-test.ts | 18 +++---- 3 files changed, 70 insertions(+), 52 deletions(-) diff --git a/smp-web/src/agent/store-idb.ts b/smp-web/src/agent/store-idb.ts index c238950fe..ac6b6ca5b 100644 --- a/smp-web/src/agent/store-idb.ts +++ b/smp-web/src/agent/store-idb.ts @@ -49,8 +49,8 @@ function createSchema(db: IDBDatabase): void { db.createObjectStore("users", {keyPath: "user_id", autoIncrement: true}) // servers — agent_schema.sql:6-11 - // CREATE TABLE servers(host TEXT, port TEXT, key_hash BLOB, PRIMARY KEY(host, port)) - db.createObjectStore("servers", {keyPath: ["host", "port"]}) + // Deviation from Haskell: keyHash is part of the primary key (Haskell uses (host,port) as workaround to avoid migration) + db.createObjectStore("servers", {keyPath: ["host", "port", "key_hash"]}) // connections — agent_schema.sql:12-31 const conns = db.createObjectStore("connections", {keyPath: "conn_id"}) @@ -196,6 +196,8 @@ function createStore(db: IDBDatabase): AgentStore { user.deleted = 1 await idbReq(store.put(user)) } + const allConns = await idbReq(tx.objectStore("connections").getAll()) + return allConns.filter((c: any) => c.user_id === userId).map((c: any) => c.conn_id) }) }, @@ -208,7 +210,7 @@ function createStore(db: IDBDatabase): AgentStore { async createServer(host, port, keyHash) { return withTx("servers", "readwrite", async (tx) => { const store = tx.objectStore("servers") - const existing = await idbReq(store.get([host, port])) + const existing = await idbReq(store.get([host, port, keyHash])) if (!existing) await idbReq(store.add({host, port, key_hash: keyHash})) }) }, @@ -393,7 +395,7 @@ function createStore(db: IDBDatabase): AgentStore { async getDeletedConnIds() { return withTx("connections", "readonly", async (tx) => { const all = await idbReq(tx.objectStore("connections").getAll()) - return all.filter((c: any) => c.deleted === 1 && !c.deleted_at_wait_delivery).map((c: any) => c.conn_id) + return all.filter((c: any) => c.deleted === 1).map((c: any) => c.conn_id) }) }, @@ -406,10 +408,12 @@ function createStore(db: IDBDatabase): AgentStore { }) }, - // getConnIds — SELECT conn_id FROM connections + // getConnIds (AgentStore.hs:2245-2246) + // SQL: SELECT conn_id FROM connections WHERE deleted = 0 async getConnIds() { return withTx("connections", "readonly", async (tx) => { - return idbReq(tx.objectStore("connections").getAllKeys()) + const all = await idbReq(tx.objectStore("connections").getAll()) + return all.filter((c: any) => c.deleted === 0).map((c: any) => c.conn_id) }) }, @@ -419,11 +423,12 @@ function createStore(db: IDBDatabase): AgentStore { return withTx(["rcv_queues", "servers"], "readwrite", async (tx) => { // createServer (INSERT OR IGNORE) const srvStore = tx.objectStore("servers") - if (!(await idbReq(srvStore.get([rcvQueue.host, rcvQueue.port])))) await idbReq(srvStore.add({host: rcvQueue.host, port: rcvQueue.port, key_hash: rcvQueue.serverKeyHash})) - // Determine rcv_queue_id: SELECT MAX(rcv_queue_id) FROM rcv_queues WHERE conn_id = ? + if (!(await idbReq(srvStore.get([rcvQueue.host, rcvQueue.port, rcvQueue.serverKeyHash])))) await idbReq(srvStore.add({host: rcvQueue.host, port: rcvQueue.port, key_hash: rcvQueue.serverKeyHash})) + // Haskell: first check if queue with same (conn_id, host, port, snd_id) exists and reuse its rcv_queue_id + // SELECT rcv_queue_id FROM rcv_queues WHERE conn_id = ? AND host = ? AND port = ? AND snd_id = ? const existing = await allByIndex(tx.objectStore("rcv_queues"), {conn_id: connId}) - const maxId = existing.reduce((m: number, q: any) => Math.max(m, q.rcv_queue_id || 0), 0) - const qId = maxId + 1 + const curr = existing.find((q: any) => q.host === rcvQueue.host && q.port === rcvQueue.port && eqBytes(q.snd_id, rcvQueue.sndId)) + const qId = curr ? curr.rcv_queue_id : (existing.reduce((m: number, q: any) => Math.max(m, q.rcv_queue_id || 0), 0) + 1) const toSubscribe = subMode === "SMOnlyCreate" ? 1 : 0 await idbReq(tx.objectStore("rcv_queues").add({ host: rcvQueue.host, port: rcvQueue.port, rcv_id: rcvQueue.rcvId, @@ -444,10 +449,12 @@ function createStore(db: IDBDatabase): AgentStore { async addConnSndQueue(connId, sndQueue) { return withTx(["snd_queues", "servers"], "readwrite", async (tx) => { const srvStore2 = tx.objectStore("servers") - if (!(await idbReq(srvStore2.get([sndQueue.host, sndQueue.port])))) await idbReq(srvStore2.add({host: sndQueue.host, port: sndQueue.port, key_hash: sndQueue.serverKeyHash})) + if (!(await idbReq(srvStore2.get([sndQueue.host, sndQueue.port, sndQueue.serverKeyHash])))) await idbReq(srvStore2.add({host: sndQueue.host, port: sndQueue.port, key_hash: sndQueue.serverKeyHash})) + // Haskell: first check if queue with same (conn_id, host, port, snd_id) exists and reuse its snd_queue_id + // SELECT snd_queue_id FROM snd_queues WHERE conn_id = ? AND host = ? AND port = ? AND snd_id = ? const existing = await allByIndex(tx.objectStore("snd_queues"), {conn_id: connId}) - const maxId = existing.reduce((m: number, q: any) => Math.max(m, q.snd_queue_id || 0), 0) - const qId = maxId + 1 + const curr = existing.find((q: any) => q.host === sndQueue.host && q.port === sndQueue.port && eqBytes(q.snd_id, sndQueue.sndId)) + const qId = curr ? curr.snd_queue_id : (existing.reduce((m: number, q: any) => Math.max(m, q.snd_queue_id || 0), 0) + 1) // ON CONFLICT DO UPDATE → use put (upsert) await idbReq(tx.objectStore("snd_queues").put({ host: sndQueue.host, port: sndQueue.port, snd_id: sndQueue.sndId, @@ -589,22 +596,21 @@ function createStore(db: IDBDatabase): AgentStore { // JOIN connections c ON q.conn_id = c.conn_id // WHERE [q.to_subscribe = 1 AND] c.deleted = 0 AND q.deleted = 0 async getSubscriptionServers(onlyNeeded) { - return withTx(["rcv_queues", "connections", "servers"], "readonly", async (tx) => { + return withTx(["rcv_queues", "connections"], "readonly", async (tx) => { const allQ = await idbReq(tx.objectStore("rcv_queues").getAll()) const connStore = tx.objectStore("connections") - const srvStore = tx.objectStore("servers") const seen = new Set() - const result: Array<{userId: number, host: string, port: string}> = [] + const result: Array<{userId: number, host: string, port: string, keyHash: Uint8Array}> = [] for (const q of allQ) { if (q.deleted) continue if (onlyNeeded && !q.to_subscribe) continue + if (!q.server_key_hash) continue const conn = await idbReq(connStore.get(q.conn_id)) if (!conn || conn.deleted !== 0) continue - const keyHash = q.server_key_hash ?? (await idbReq(srvStore.get([q.host, q.port])))?.key_hash - const key = `${conn.user_id}:${q.host}:${q.port}:${keyHash ? toHex(keyHash) : ""}` + const key = `${conn.user_id}:${q.host}:${q.port}:${toHex(q.server_key_hash)}` if (!seen.has(key)) { seen.add(key) - result.push({userId: conn.user_id, host: q.host, port: q.port}) + result.push({userId: conn.user_id, host: q.host, port: q.port, keyHash: q.server_key_hash}) } } return result @@ -615,16 +621,16 @@ function createStore(db: IDBDatabase): AgentStore { // SQL: rcvQueueSubQuery WHERE [q.to_subscribe = 1 AND] c.deleted = 0 AND q.deleted = 0 // AND c.user_id = ? AND q.host = ? AND q.port = ? AND COALESCE(q.server_key_hash, s.key_hash) = ? // ORDER BY q.rcv_id LIMIT ? - async getUserServerRcvQueueSubs(userId, host, port, onlyNeeded, batchSize, cursor) { - return withTx(["rcv_queues", "connections", "servers"], "readonly", async (tx) => { + async getUserServerRcvQueueSubs(userId, host, port, keyHash, onlyNeeded, batchSize, cursor) { + return withTx(["rcv_queues", "connections"], "readonly", async (tx) => { const allQ = await idbReq(tx.objectStore("rcv_queues").getAll()) const connStore = tx.objectStore("connections") - const srvStore = tx.objectStore("servers") // Filter matching queues const matching: any[] = [] for (const q of allQ) { if (q.deleted) continue if (q.host !== host || q.port !== port) continue + if (!q.server_key_hash || !eqBytes(q.server_key_hash, keyHash)) continue if (onlyNeeded && !q.to_subscribe) continue if (cursor !== null && compareUint8Array(q.rcv_id, cursor as any) <= 0) continue const conn = await idbReq(connStore.get(q.conn_id)) @@ -1145,8 +1151,8 @@ function createStore(db: IDBDatabase): AgentStore { // JOIN messages const msg = await idbReq(tx.objectStore("messages").get([connId, internalId])) if (!msg) return null - // LEFT JOIN snd_messages ON rcpt_internal_id = r.internal_id - const sndAll = await allByIndex(tx.objectStore("snd_messages"), {conn_id: connId, internal_id: internalId}) + // LEFT JOIN snd_messages s ON s.conn_id = r.conn_id AND s.rcpt_internal_id = r.internal_id + const sndAll = await allByIndex(tx.objectStore("snd_messages"), {conn_id: connId}) const sndRcpt = sndAll.find((s: any) => s.rcpt_internal_id === internalId) return { internalId: rm.internal_id, @@ -1183,6 +1189,9 @@ function createStore(db: IDBDatabase): AgentStore { // JOIN messages const msg = await idbReq(tx.objectStore("messages").get([connId, rm.internal_id])) if (!msg) return null + // LEFT JOIN snd_messages s ON s.conn_id = r.conn_id AND s.rcpt_internal_id = r.internal_id + const sndAll = await allByIndex(tx.objectStore("snd_messages"), {conn_id: connId}) + const sndRcpt = sndAll.find((s: any) => s.rcpt_internal_id === rm.internal_id) return { internalId: rm.internal_id, msgMeta: { @@ -1195,7 +1204,7 @@ function createStore(db: IDBDatabase): AgentStore { msgType: msg.msg_type, msgBody: msg.msg_body, userAck: rm.user_ack === 1, - msgReceipt: null, + msgReceipt: sndRcpt?.rcpt_status ?? null, } }) }, @@ -1327,7 +1336,9 @@ function createStore(db: IDBDatabase): AgentStore { internalId: sm.internal_id, msgType: msg.msg_type, internalHash: sm.internal_hash, - msgReceipt: sm.rcpt_status ?? null, + msgReceipt: sm.rcpt_internal_id != null && sm.rcpt_status != null + ? {agentMsgId: sm.rcpt_internal_id, msgRcptStatus: sm.rcpt_status} + : null, } }) }, @@ -1339,8 +1350,8 @@ function createStore(db: IDBDatabase): AgentStore { const store = tx.objectStore("snd_messages") const sm = await idbReq(store.get([connId, sndMsgId])) if (sm) { - sm.rcpt_internal_id = receipt - sm.rcpt_status = receipt + sm.rcpt_internal_id = receipt.agentMsgId + sm.rcpt_status = receipt.msgRcptStatus await idbReq(store.put(sm)) } }) @@ -1536,17 +1547,24 @@ function createStore(db: IDBDatabase): AgentStore { }, // getPendingServerCommand (AgentStore.hs:1439-1480) - // getCmdId SQL: SELECT command_id FROM commands - // WHERE conn_id = ? AND host = ? AND port = ? AND failed = 0 - // ORDER BY created_at ASC, command_id ASC LIMIT 1 + // When host/port are null: + // SQL: SELECT command_id FROM commands WHERE conn_id = ? AND host IS NULL AND port IS NULL AND failed = 0 + // ORDER BY created_at ASC, command_id ASC LIMIT 1 + // When host/port are provided: + // SQL: SELECT command_id FROM commands WHERE conn_id = ? AND host = ? AND port = ? AND failed = 0 + // ORDER BY created_at ASC, command_id ASC LIMIT 1 // getCommand SQL: SELECT c.corr_id, cs.user_id, c.command FROM commands c // JOIN connections cs USING (conn_id) WHERE c.command_id = ? - async getPendingServerCommand(host, port) { + async getPendingServerCommand(connId, host, port) { return withTx(["commands", "connections"], "readonly", async (tx) => { - const allCmds = await allByIndex(tx.objectStore("commands"), {host, port}) - // Filter non-failed, sort by created_at then command_id + const allCmds = await allByIndex(tx.objectStore("commands"), {conn_id: connId}) + // Filter by host/port and non-failed const pending = allCmds - .filter((c: any) => !c.failed) + .filter((c: any) => { + if (c.failed) return false + if (host === null && port === null) return c.host == null && c.port == null + return c.host === host && c.port === port + }) .sort((a: any, b: any) => { const tsCompare = (a.created_at || "").localeCompare(b.created_at || "") return tsCompare !== 0 ? tsCompare : (a.command_id - b.command_id) diff --git a/smp-web/src/agent/store.ts b/smp-web/src/agent/store.ts index 8aa65e4ed..fc14f1447 100644 --- a/smp-web/src/agent/store.ts +++ b/smp-web/src/agent/store.ts @@ -134,7 +134,7 @@ export interface RcvMsg { msgType: string msgBody: Uint8Array userAck: boolean - msgReceipt: Uint8Array | null + msgReceipt: {agentMsgId: number, msgRcptStatus: string} | null } export interface PendingQueueMsg { @@ -173,7 +173,7 @@ export interface AgentStore { createUserRecord(): Promise getUserIds(): Promise deleteUserRecord(userId: UserId): Promise - setUserDeleted(userId: UserId): Promise + setUserDeleted(userId: UserId): Promise // -- Servers (AgentStore.hs:233-240) createServer(host: string, port: string, keyHash: Uint8Array): Promise @@ -213,8 +213,8 @@ export interface AgentStore { setConnectionNtfs(connId: ConnId, enable: boolean): Promise // -- Subscriptions (AgentStore.hs:700-800) - getSubscriptionServers(onlyNeeded: boolean): Promise> - getUserServerRcvQueueSubs(userId: UserId, host: string, port: string, onlyNeeded: boolean, batchSize: number, cursor: number | null): Promise<{queues: RcvQueue[], nextCursor: number | null}> + getSubscriptionServers(onlyNeeded: boolean): Promise> + getUserServerRcvQueueSubs(userId: UserId, host: string, port: string, keyHash: Uint8Array, onlyNeeded: boolean, batchSize: number, cursor: number | null): Promise<{queues: RcvQueue[], nextCursor: number | null}> unsetQueuesToSubscribe(): Promise getConnectionsForDelivery(): Promise getAllSndQueuesForDelivery(): Promise @@ -251,10 +251,10 @@ export interface AgentStore { checkRcvMsgHashExists(connId: ConnId, hash: Uint8Array): Promise getRcvMsgBrokerTs(connId: ConnId, brokerId: Uint8Array): Promise deleteMsg(connId: ConnId, internalId: number): Promise - deleteDeliveredSndMsg(connId: ConnId, internalSndId: number): Promise + deleteDeliveredSndMsg(connId: ConnId, internalId: number): Promise deleteSndMsgDelivery(connId: ConnId, sndQueue: SndQueue, msgId: number, keepForReceipt: boolean): Promise - getSndMsgViaRcpt(connId: ConnId, sndMsgId: number): Promise<{internalId: number, msgType: string, internalHash: Uint8Array, msgReceipt: Uint8Array | null} | null> - updateSndMsgRcpt(connId: ConnId, sndMsgId: number, receipt: Uint8Array): Promise + getSndMsgViaRcpt(connId: ConnId, sndMsgId: number): Promise<{internalId: number, msgType: string, internalHash: Uint8Array, msgReceipt: {agentMsgId: number, msgRcptStatus: string} | null} | null> + updateSndMsgRcpt(connId: ConnId, sndMsgId: number, receipt: {agentMsgId: number, msgRcptStatus: string}): Promise // -- Ratchet (AgentStore.hs:1300-1400) createRatchetX3dhKeys(connId: ConnId, privKey1: Uint8Array, privKey2: Uint8Array, pqKem: Uint8Array | null): Promise @@ -269,7 +269,7 @@ export interface AgentStore { createCommand(corrId: Uint8Array, connId: ConnId, host: string | null, port: string | null, command: AsyncCommand): Promise getPendingCommandServers(connIds: ConnId[]): Promise> getAllPendingCommandConns(): Promise> - getPendingServerCommand(host: string, port: string): Promise + getPendingServerCommand(connId: ConnId, host: string | null, port: string | null): Promise updateCommandServer(commandId: number, host: string, port: string): Promise deleteCommand(commandId: number): Promise diff --git a/smp-web/tests/store-test.ts b/smp-web/tests/store-test.ts index d97631e29..a8a0ee6a1 100644 --- a/smp-web/tests/store-test.ts +++ b/smp-web/tests/store-test.ts @@ -303,7 +303,8 @@ async function testSubscriptions(store: AgentStore) { lastInternalMsgId: 0, lastInternalRcvMsgId: 0, lastInternalSndMsgId: 0, lastExternalSndMsgId: 0, lastRcvMsgHash: new Uint8Array(0), lastSndMsgHash: new Uint8Array(0), }, "INV") - await store.createServer("sub.example.com", "5223", randomBytes(32)) + const subKeyHash = randomBytes(32) + await store.createServer("sub.example.com", "5223", subKeyHash) const rcvQ = await store.addConnRcvQueue(connId, { host: "sub.example.com", port: "5223", rcvId: randomBytes(24), connId, rcvPrivateKey: randomBytes(32), rcvDhSecret: randomBytes(32), @@ -311,7 +312,7 @@ async function testSubscriptions(store: AgentStore) { sndId: randomBytes(24), sndKey: null, status: "new" as const, smpClientVersion: 7, dbQueueId: 0, primary: true, replaceRcvQueueId: null, queueMode: null, - serverKeyHash: randomBytes(32), lastBrokerTs: null, + serverKeyHash: subKeyHash, lastBrokerTs: null, }, "SMOnlyCreate") // getSubscriptionServers — onlyNeeded=true should find our to_subscribe=1 queue @@ -323,7 +324,7 @@ async function testSubscriptions(store: AgentStore) { assert(allSrvs.some(s => s.host === "sub.example.com"), "getSubscriptionServers(false) finds all") // getUserServerRcvQueueSubs - const {queues} = await store.getUserServerRcvQueueSubs(userId, "sub.example.com", "5223", true, 10, null) + const {queues} = await store.getUserServerRcvQueueSubs(userId, "sub.example.com", "5223", subKeyHash, true, 10, null) assert(queues.length >= 1, "getUserServerRcvQueueSubs finds queues") // unsetQueuesToSubscribe @@ -612,8 +613,7 @@ async function testSendMessages(store: AgentStore) { assertEq(sndMsg!.internalHash, internalHash, "getSndMsgViaRcpt hash matches") // updateSndMsgRcpt - const receipt = randomBytes(8) - await store.updateSndMsgRcpt(connId, internalSndId, receipt) + await store.updateSndMsgRcpt(connId, internalSndId, {agentMsgId: internalId, msgRcptStatus: "ok"}) // deleteSndMsgDelivery — deletes delivery, then msg if no more deliveries await store.deleteSndMsgDelivery(connId, sndQueue, internalId, false) @@ -747,19 +747,19 @@ async function testCommands(store: AgentStore) { assert(allConns.some(c => hex(c.connId) === hex(connId)), "getAllPendingCommandConns finds conn") // getPendingServerCommand - const pendingCmd = await store.getPendingServerCommand("cmd.example.com", "5223") + const pendingCmd = await store.getPendingServerCommand(connId, "cmd.example.com", "5223") assert(pendingCmd !== null, "getPendingServerCommand finds command") // updateCommandServer await store.updateCommandServer(cmdId, "cmd2.example.com", "5224") - const updated = await store.getPendingServerCommand("cmd2.example.com", "5224") + const updated = await store.getPendingServerCommand(connId, "cmd2.example.com", "5224") assert(updated !== null, "updateCommandServer changes server") - const oldHost = await store.getPendingServerCommand("cmd.example.com", "5223") + const oldHost = await store.getPendingServerCommand(connId, "cmd.example.com", "5223") assert(oldHost === null, "updateCommandServer — old host returns nothing") // deleteCommand await store.deleteCommand(cmdId) - const deleted = await store.getPendingServerCommand("cmd2.example.com", "5224") + const deleted = await store.getPendingServerCommand(connId, "cmd2.example.com", "5224") assert(deleted === null, "deleteCommand removes command") }