From 4e94a7701925f74d076b42437a884547a8700d3f Mon Sep 17 00:00:00 2001 From: ht Date: Mon, 14 Sep 2026 21:46:52 +0300 Subject: [PATCH] fix: prevent offline queue replay data loss --- src/context/AppContext.jsx | 1 + src/services/offlineMutationQueue.js | 48 +++++++++++++++++++++++---- tests/offline-mutation-queue.test.mjs | 46 +++++++++++++++++++++++++ 3 files changed, 88 insertions(+), 7 deletions(-) diff --git a/src/context/AppContext.jsx b/src/context/AppContext.jsx index 1555a5f..8fbc51e 100644 --- a/src/context/AppContext.jsx +++ b/src/context/AppContext.jsx @@ -221,6 +221,7 @@ export const AppProvider = ({ children }) => { if (getStoredUser()?.email !== owner) throw new Error('Oturum değişti; kuyruk bekletildi.'); savedTenderStates.current = { ...savedTenderStates.current, [item.key]: record }; setTenders(prev => applyTenderStates(prev, savedTenderStates.current)); + return record; }); setOfflineQueueSize(offlineMutationCount(owner)); if (result.error) setSaveError(`Çevrimdışı kayıt gönderilemedi: ${result.error.message}`); diff --git a/src/services/offlineMutationQueue.js b/src/services/offlineMutationQueue.js index 90085aa..b6d090d 100644 --- a/src/services/offlineMutationQueue.js +++ b/src/services/offlineMutationQueue.js @@ -1,5 +1,6 @@ const PREFIX = 'ihaleist_offline_mutations:'; const LIMIT = 100; +const activeReplays = new Map(); const ownerKey = owner => `${PREFIX}${encodeURIComponent(String(owner || '').trim().toLowerCase())}`; @@ -14,7 +15,7 @@ function readStorage(owner) { function writeStorage(owner, rows) { if (typeof localStorage === 'undefined' || !String(owner || '').trim()) return false; try { - localStorage.setItem(ownerKey(owner), JSON.stringify(rows.slice(-LIMIT))); + localStorage.setItem(ownerKey(owner), JSON.stringify(rows)); return true; } catch { return false; } } @@ -39,6 +40,9 @@ export function enqueueTenderMutation(owner, { key, patch, snapshot, version }) const rows = readStorage(owner); const index = rows.findIndex(item => item.kind === 'tender-state' && item.key === key); const previous = index >= 0 ? rows[index] : null; + if (!previous && rows.length >= LIMIT) { + throw new Error('Çevrimdışı kayıt kuyruğu dolu. Bağlantı kurulduktan sonra yeniden deneyin.'); + } const next = { id: previous?.id || `offline-${crypto.randomUUID()}`, kind: 'tender-state', @@ -49,6 +53,7 @@ export function enqueueTenderMutation(owner, { key, patch, snapshot, version }) createdAt: previous?.createdAt || new Date().toISOString(), updatedAt: new Date().toISOString(), attempts: previous?.attempts || 0, + revision: (Number.isInteger(previous?.revision) ? previous.revision : 0) + 1, }; if (index >= 0) rows[index] = next; else rows.push(next); if (!writeStorage(owner, rows)) throw new Error('Çevrimdışı kayıt kuyruğu yazılamadı.'); @@ -67,27 +72,56 @@ export function markOfflineMutationAttempt(owner, id) { return next.find(item => item.id === id) || null; } -/** Replay in order; a failed item remains queued and stops the batch. */ -export async function replayOfflineMutations(owner, send) { +function acknowledgeOfflineMutation(owner, item, record) { + const rows = readStorage(owner); + const index = rows.findIndex(row => row.id === item.id); + if (index < 0) return true; + const current = rows[index]; + if ((current.revision || 0) === (item.revision || 0)) { + rows.splice(index, 1); + } else { + const nextVersion = Number(record?.version); + rows[index] = { + ...current, + version: Number.isInteger(nextVersion) && nextVersion >= 0 ? nextVersion : current.version, + }; + } + return writeStorage(owner, rows); +} + +async function runReplay(owner, send) { const rows = listOfflineMutations(owner); const result = { sent: 0, pending: rows.length, error: null }; for (const item of rows) { markOfflineMutationAttempt(owner, item.id); try { - await send(item); - removeOfflineMutation(owner, item.id); + const record = await send(item); + if (!acknowledgeOfflineMutation(owner, item, record)) { + throw new Error('Çevrimdışı kayıt kuyruğu güncellenemedi.'); + } result.sent += 1; - result.pending -= 1; } catch (error) { result.error = error; break; } } + result.pending = offlineMutationCount(owner); return result; } +/** Replay in order; concurrent reconnect signals share one replay operation. */ +export function replayOfflineMutations(owner, send) { + const replayKey = ownerKey(owner); + const active = activeReplays.get(replayKey); + if (active) return active; + const replay = runReplay(owner, send).finally(() => { + if (activeReplays.get(replayKey) === replay) activeReplays.delete(replayKey); + }); + activeReplays.set(replayKey, replay); + return replay; +} + export function clearOfflineMutations(owner) { if (typeof localStorage === 'undefined') return false; try { localStorage.removeItem(ownerKey(owner)); return true; } catch { return false; } } - diff --git a/tests/offline-mutation-queue.test.mjs b/tests/offline-mutation-queue.test.mjs index 0502c4d..f60c7cf 100644 --- a/tests/offline-mutation-queue.test.mjs +++ b/tests/offline-mutation-queue.test.mjs @@ -43,3 +43,49 @@ test('replay removes acknowledged rows and retains the first failed row', async clearOfflineMutations('replay@example.test'); }); +test('concurrent reconnect signals send each queued mutation once', async () => { + clearOfflineMutations('concurrent@example.test'); + enqueueTenderMutation('concurrent@example.test', { key: 'a', patch: { takipte: true }, snapshot: {}, version: 0 }); + let sends = 0; + let release; + const gate = new Promise(resolve => { release = resolve; }); + const send = async () => { sends += 1; await gate; return { version: 1 }; }; + const first = replayOfflineMutations('concurrent@example.test', send); + const second = replayOfflineMutations('concurrent@example.test', send); + release(); + const [a, b] = await Promise.all([first, second]); + assert.equal(sends, 1); + assert.deepEqual(a, b); + assert.equal(offlineMutationCount('concurrent@example.test'), 0); +}); + +test('a mutation added during replay is retained with the acknowledged server version', async () => { + clearOfflineMutations('during-replay@example.test'); + enqueueTenderMutation('during-replay@example.test', { key: 'a', patch: { takipte: true }, snapshot: {}, version: 3 }); + const first = await replayOfflineMutations('during-replay@example.test', async item => { + enqueueTenderMutation('during-replay@example.test', { key: 'a', patch: { kanbanStatus: 'teklif_hazirlaniyor' }, snapshot: {}, version: item.version }); + return { version: 4 }; + }); + assert.equal(first.pending, 1); + const retained = listOfflineMutations('during-replay@example.test')[0]; + assert.equal(retained.version, 4); + assert.deepEqual(retained.patch, { takipte: true, kanbanStatus: 'teklif_hazirlaniyor' }); + const versions = []; + await replayOfflineMutations('during-replay@example.test', async item => { versions.push(item.version); return { version: 5 }; }); + assert.deepEqual(versions, [4]); + assert.equal(offlineMutationCount('during-replay@example.test'), 0); +}); + +test('a full queue rejects a new tender instead of silently discarding an older change', () => { + clearOfflineMutations('full@example.test'); + for (let index = 0; index < 100; index += 1) { + enqueueTenderMutation('full@example.test', { key: String(index), patch: { takipte: true }, snapshot: {}, version: 0 }); + } + assert.throws( + () => enqueueTenderMutation('full@example.test', { key: 'overflow', patch: { takipte: true }, snapshot: {}, version: 0 }), + /kuyruğu dolu/, + ); + assert.equal(offlineMutationCount('full@example.test'), 100); + assert.equal(listOfflineMutations('full@example.test')[0].key, '0'); + clearOfflineMutations('full@example.test'); +});