diff --git a/docs/SYNC-merge-vectors.json b/docs/SYNC-merge-vectors.json index 36e0382..1f8e342 100644 --- a/docs/SYNC-merge-vectors.json +++ b/docs/SYNC-merge-vectors.json @@ -835,5 +835,494 @@ } } } + ], + "channels": [ + { + "what": "a stamped field held here beats an old unstamped copy arriving", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Renamed", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 5000 + } + } + ] + }, + "incoming": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [] + } + ] + }, + "expected": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Renamed", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 5000 + } + } + ], + "channelsLeftAt": {} + } + }, + { + "what": "a stamped field arriving beats the old unstamped copy held here", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [] + } + ] + }, + "incoming": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Renamed", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 5000 + } + } + ] + }, + "expected": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Renamed", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 5000 + } + } + ], + "channelsLeftAt": {} + } + }, + { + "what": "two devices changing different fields keep both changes", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Renamed", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 5000 + } + } + ] + }, + "incoming": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [ + "0x00000000000000000000000000000000000000b1" + ], + "fieldTs": { + "accessSnapshot": 6000 + } + } + ] + }, + "expected": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Renamed", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [ + "0x00000000000000000000000000000000000000b1" + ], + "fieldTs": { + "accessSnapshot": 6000, + "name": 5000 + } + } + ], + "channelsLeftAt": {} + } + }, + { + "what": "the same two changes merged in the other order give the same record", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [ + "0x00000000000000000000000000000000000000b1" + ], + "fieldTs": { + "accessSnapshot": 6000 + } + } + ] + }, + "incoming": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Renamed", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 5000 + } + } + ] + }, + "expected": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Renamed", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [ + "0x00000000000000000000000000000000000000b1" + ], + "fieldTs": { + "name": 5000, + "accessSnapshot": 6000 + } + } + ], + "channelsLeftAt": {} + } + }, + { + "what": "a client that drops the stamps neither reverts a stamped field nor erases the stamps", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 90, + "accessSnapshot": [], + "fieldTs": { + "storageDays": 5000 + } + } + ] + }, + "incoming": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [] + } + ] + }, + "expected": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 90, + "accessSnapshot": [], + "fieldTs": { + "storageDays": 5000 + } + } + ], + "channelsLeftAt": {} + } + }, + { + "what": "an older stamp arriving does not undo a newer one held here", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Newer", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 6000 + } + } + ] + }, + "incoming": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Older", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 5000 + } + } + ] + }, + "expected": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Newer", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 6000 + } + } + ], + "channelsLeftAt": {} + } + }, + { + "what": "equal stamps fall back to the join-time rule, which gives the tie to the incoming copy", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Here", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 5000 + } + } + ] + }, + "incoming": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Arriving", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 5000 + } + } + ] + }, + "expected": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Arriving", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 5000 + } + } + ], + "channelsLeftAt": {} + } + }, + { + "what": "a field only one copy carries is kept, whichever copy wins the others", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "keysStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-4", + "storageProvider": "streamr" + } + ] + }, + "incoming": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Renamed", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "inboxStreamId": null, + "fieldTs": { + "name": 5000 + } + } + ] + }, + "expected": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Renamed", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "inboxStreamId": null, + "keysStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-4", + "storageProvider": "streamr", + "fieldTs": { + "name": 5000 + } + } + ], + "channelsLeftAt": {} + } + }, + { + "what": "a leave newer than the join removes the channel whatever its stamps say", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Renamed", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 9000 + } + } + ] + }, + "incoming": { + "channels": [], + "channelsLeftAt": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": 5000 + } + }, + "expected": { + "channels": [], + "channelsLeftAt": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": 5000 + } + } + }, + { + "what": "a join newer than the leave keeps the channel and retires the leave", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 7000, + "storageDays": 30, + "accessSnapshot": [] + } + ] + }, + "incoming": { + "channels": [], + "channelsLeftAt": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": 5000 + } + }, + "expected": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 7000, + "storageDays": 30, + "accessSnapshot": [] + } + ], + "channelsLeftAt": {} + } + } ] } diff --git a/src/js/channels.js b/src/js/channels.js index 741b4ac..fdfefd4 100644 --- a/src/js/channels.js +++ b/src/js/channels.js @@ -40,12 +40,17 @@ import { DeliveryConfirm } from './channels/DeliveryConfirm.js'; import { Membership } from './channels/Membership.js'; import { RotationRetry } from './channels/RotationRetry.js'; import { ModDeltas, MOD_ACTION_TYPE } from './channels/ModDeltas.js'; +import { stampChangedFields } from './syncMerge.js'; const GATE_REPAIR_WAIT_MS = 10_000; class ChannelManager { constructor() { this.channels = new Map(); // streamId -> channel object + // Each record as last persisted or imported, serialized: what a save + // compares against to stamp the fields that changed here. A copy, as + // the live records share their arrays with the storage cache. + this._persisted = new Map(); this.currentChannel = null; // Gated PREVIEW shadow (N-D): a token/NFT holder browsing from // Explore. The preview lives outside `channels` on purpose (it must @@ -282,6 +287,7 @@ class ChannelManager { } else { Logger.debug('No saved channels found'); } + this._rememberPersisted(); } /** @@ -412,6 +418,7 @@ class ChannelManager { } this.channels = nextChannels; + this._rememberPersisted(); Logger.debug(`Reloaded ${channelsData.length} channels from sync snapshot (authoritative)`); @@ -421,6 +428,77 @@ class ChannelManager { }; } + /** + * A channel as persisted: metadata only. Messages and reactions are + * loaded from storage on demand; adminState is rebuilt from -3/P0 on + * subscribe. + */ + _storedRecord(ch) { + return { + messageStreamId: ch.messageStreamId, + ephemeralStreamId: ch.ephemeralStreamId, + adminStreamId: ch.adminStreamId, + name: ch.name, + type: ch.type, + // Gated (N-C): without the gate address every gated code path + // silently degrades after a reload — the publish falls back to + // an ephemeral key the network rejects (MISSING_PERMISSION). + gate: ch.gate || null, + // Author visibility — losing it would flip a Sealed + // channel back to clone publishes (account on the wire). + wireIdentity: ch.wireIdentity || null, + createdAt: ch.createdAt, + createdBy: ch.createdBy, + // Local membership timestamp — drives per-channel latest-wins + // in cross-device sync (join vs leave tombstone comparison) + joinedAt: ch.joinedAt || ch.createdAt || null, + password: ch.password, + members: ch.members || [], + // Access losses this device has already rotated the epoch for; + // without it every admin open would rotate again for the same + // cut. (rotatedForBanned is the older, narrower name of the same set.) + rotatedForNoAccess: ch.rotatedForNoAccess || ch.rotatedForBanned || [], + // Who had gate access at the last sweep — losing it is what + // triggers the deferred rotation. + accessSnapshot: ch.accessSnapshot || [], + // Addresses banned from here, kept as gate-read candidates so + // Moderation can still list them after a reload. + knownBanned: ch.knownBanned || [], + storageEnabled: ch.storageEnabled, + // Retention per stored stream, last read off-chain. Fallbacks + // for when the Graph is unreachable on a later open, and the + // only value the headless epoch-key sweep can consult: unsaved, + // every reload reverts them to the 180-day default and disarms + // both the TTL republish and the key re-announce. + storageDays: ch.storageDays ?? null, + adminStorageDays: ch.adminStorageDays ?? null, + keysStorageDays: ch.keysStorageDays ?? null, + interactionsStorageDays: ch.interactionsStorageDays ?? null, + // Exposure and metadata + exposure: ch.exposure || 'hidden', + description: ch.description || '', + language: ch.language || '', + category: ch.category || '', + // Timestamp of the last local on-chain metadata edit (name/description) + // — prevents Graph indexing lag from reverting local admin edits + metaUpdatedAt: ch.metaUpdatedAt || null, + // Channel options + readOnly: ch.readOnly || false, + writeOnly: ch.writeOnly || false, + classification: ch.classification || null, + // DM-specific + peerAddress: ch.peerAddress || null, + inboxStreamId: ch.inboxStreamId || null, + // When each field last changed on some device; the sync merges + // the record field by field on it. + ...(ch.fieldTs ? { fieldTs: ch.fieldTs } : {}) + }; + } + + _rememberPersisted(records = Array.from(this.channels.values()).map(ch => this._storedRecord(ch))) { + this._persisted = new Map(records.map(r => [r.messageStreamId, JSON.parse(JSON.stringify(r))])); + } + /** * Save channels to secure storage (encrypted) * NOTE: messages and reactions are NOT persisted - they come from storage @@ -431,68 +509,16 @@ class ChannelManager { Logger.warn('Secure storage not unlocked - cannot save channels'); return; } - - // Strip messages and reactions from persistence - only save metadata - // Messages are loaded from storage on demand (lazy loading) - const channelsData = Array.from(this.channels.values()).map(ch => ({ - messageStreamId: ch.messageStreamId, - ephemeralStreamId: ch.ephemeralStreamId, - adminStreamId: ch.adminStreamId, - name: ch.name, - type: ch.type, - // Gated (N-C): without the gate address every gated code path - // silently degrades after a reload — the publish falls back to - // an ephemeral key the network rejects (MISSING_PERMISSION). - gate: ch.gate || null, - // Author visibility — losing it would flip a Sealed - // channel back to clone publishes (account on the wire). - wireIdentity: ch.wireIdentity || null, - createdAt: ch.createdAt, - createdBy: ch.createdBy, - // Local membership timestamp — drives per-channel latest-wins - // in cross-device sync (join vs leave tombstone comparison) - joinedAt: ch.joinedAt || ch.createdAt || null, - password: ch.password, - members: ch.members || [], - // Access losses this device has already rotated the epoch for; - // without it every admin open would rotate again for the same - // cut. (rotatedForBanned is the older, narrower name of the same set.) - rotatedForNoAccess: ch.rotatedForNoAccess || ch.rotatedForBanned || [], - // Who had gate access at the last sweep — losing it is what - // triggers the deferred rotation. - accessSnapshot: ch.accessSnapshot || [], - // Addresses banned from here, kept as gate-read candidates so - // Moderation can still list them after a reload. - knownBanned: ch.knownBanned || [], - storageEnabled: ch.storageEnabled, - // Retention per stored stream, last read off-chain. Fallbacks - // for when the Graph is unreachable on a later open, and the - // only value the headless epoch-key sweep can consult: unsaved, - // every reload reverts them to the 180-day default and disarms - // both the TTL republish and the key re-announce. - storageDays: ch.storageDays ?? null, - adminStorageDays: ch.adminStorageDays ?? null, - keysStorageDays: ch.keysStorageDays ?? null, - interactionsStorageDays: ch.interactionsStorageDays ?? null, - // Exposure and metadata - exposure: ch.exposure || 'hidden', - description: ch.description || '', - language: ch.language || '', - category: ch.category || '', - // Timestamp of the last local on-chain metadata edit (name/description) - // — prevents Graph indexing lag from reverting local admin edits - metaUpdatedAt: ch.metaUpdatedAt || null, - // Channel options - readOnly: ch.readOnly || false, - writeOnly: ch.writeOnly || false, - classification: ch.classification || null, - // DM-specific - peerAddress: ch.peerAddress || null, - inboxStreamId: ch.inboxStreamId || null - // messages: excluded - loaded from storage - // reactions: excluded - loaded from storage - // adminState: excluded - rebuilt from -3/P0 on subscribe - })); + + const channelsData = Array.from(this.channels.values()).map(ch => this._storedRecord(ch)); + const now = Date.now(); + for (const record of channelsData) { + const stamps = stampChangedFields(this._persisted.get(record.messageStreamId), record, now); + if (stamps === record.fieldTs) continue; + record.fieldTs = stamps; + this.channels.get(record.messageStreamId).fieldTs = stamps; + } + this._rememberPersisted(channelsData); Logger.debug('Saving channels to secure storage:', channelsData.length); await secureStorage.setChannels(channelsData); @@ -515,6 +541,7 @@ class ChannelManager { */ clearChannels() { this.channels.clear(); + this._persisted = new Map(); this.setCurrentChannel(null); this.onlineUsers.clear(); this.processingMessages.clear(); diff --git a/src/js/channels/MessageFlow.js b/src/js/channels/MessageFlow.js index dc11e60..8dc3b39 100644 --- a/src/js/channels/MessageFlow.js +++ b/src/js/channels/MessageFlow.js @@ -125,16 +125,6 @@ export class MessageFlow { // it via the cache stale-guard. timestamp: data.timestamp || null }); - } else if (data.type === 'member_update') { - const channel = this.manager.channels.get(streamId); - if (channel) { - channel.members = data.members; - try { - await this.manager.saveChannels(); - } catch (e) { - Logger.error('Failed to persist member update:', e); - } - } } } diff --git a/src/js/syncMerge.js b/src/js/syncMerge.js index 005dd33..292f7ae 100644 --- a/src/js/syncMerge.js +++ b/src/js/syncMerge.js @@ -108,6 +108,65 @@ export function mergeSentReactions(local, remote) { return result; } +const fieldStamp = (record, key) => { + const ts = record?.fieldTs?.[key]; + return typeof ts === 'number' ? ts : 0; +}; + +/** + * A channel record's field stamps after a local save: each field that differs + * from the copy last persisted or imported is stamped `now`. A record with no + * such copy (created here) keeps the stamps it has. + * + * @param {Object|undefined} previous - The copy last persisted or imported + * @param {Object} record - The record about to be persisted + * @param {number} now + * @returns {Object|undefined} `record.fieldTs` itself when nothing changed + */ +export function stampChangedFields(previous, record, now) { + if (!previous) return record.fieldTs; + let stamps = record.fieldTs; + for (const key of new Set([...Object.keys(record), ...Object.keys(previous)])) { + if (key === 'fieldTs' || JSON.stringify(record[key]) === JSON.stringify(previous[key])) continue; + if (stamps === record.fieldTs) stamps = { ...(record.fieldTs || {}) }; + stamps[key] = now; + } + return stamps; +} + +/** + * One channel record from two copies of it. Each field comes from the copy + * whose `fieldTs` stamped it later; a field neither copy stamped, or both + * stamped at the same time, comes from `preferred`, and a field only one copy + * carries is kept. The stamps join, the latest per field, so a client that + * drops them cannot take them from the others. + * + * @param {Object} preferred - The copy the join-time rule picks + * @param {Object} other - The other copy + * @returns {Object} `preferred` itself when nothing comes from `other` + */ +export function mergeChannelRecord(preferred, other) { + const merged = {}; + let tookOther = false; + for (const key of new Set([...Object.keys(preferred), ...Object.keys(other)])) { + if (key === 'fieldTs') continue; + const fromOther = key in other + && (!(key in preferred) || fieldStamp(other, key) > fieldStamp(preferred, key)); + merged[key] = fromOther ? other[key] : preferred[key]; + if (fromOther) tookOther = true; + } + if (!tookOther) return preferred; + if (preferred.fieldTs || other.fieldTs) { + merged.fieldTs = {}; + for (const source of [preferred.fieldTs || {}, other.fieldTs || {}]) { + for (const [key, ts] of Object.entries(source)) { + if (typeof ts === 'number' && ts > (merged.fieldTs[key] || 0)) merged.fieldTs[key] = ts; + } + } + } + return merged; +} + /** * Merge channel lists as an LWW-element-set (per-channel latest-wins). * @@ -116,6 +175,7 @@ export function mergeSentReactions(local, remote) { * `channelsLeftAt`. The most recent action wins; on a tie, Join wins. * This replaces whole-array snapshot replacement, which could delete a * fresh local Join when an older remote snapshot arrived (or vice versa). + * Two copies of the same channel merge field by field (mergeChannelRecord). * * @param {Array} baseChannels - Base channel entries * @param {Array} incomingChannels - Incoming channel entries @@ -137,8 +197,8 @@ export function mergeChannels(baseChannels, incomingChannels, baseLeftAt, incomi } } - // Union channel entries: entry with the newest join timestamp wins; - // incoming wins ties (newer snapshot has fresher metadata). An entry with + // Union channel entries: for the fields no stamp decides, the entry with + // the newest join timestamp wins and incoming wins ties. An entry with // no joinedAt of its own is never a newer join than one that has it, but // its time still counts as a join against a leave tombstone. const replaces = (incoming, existing) => { @@ -158,8 +218,12 @@ export function mergeChannels(baseChannels, incomingChannels, baseLeftAt, incomi if (!channel?.messageStreamId) continue; noteJoin(channel); const existing = byId.get(channel.messageStreamId); - if (!existing || replaces(channel, existing)) { + if (!existing) { byId.set(channel.messageStreamId, channel); + } else { + byId.set(channel.messageStreamId, replaces(channel, existing) + ? mergeChannelRecord(channel, existing) + : mergeChannelRecord(existing, channel)); } } diff --git a/tests/unit/channels.test.js b/tests/unit/channels.test.js index 54248b7..dc4fffc 100644 --- a/tests/unit/channels.test.js +++ b/tests/unit/channels.test.js @@ -1198,6 +1198,74 @@ describe('ChannelManager', () => { }); }); + // The sync merges a channel record field by field on these stamps, so a + // stamp on a field nobody changed here would let this device's stale + // value win it on every other device. + describe('saveChannels() field stamps', () => { + const stored = (extra = {}) => ({ + messageStreamId: 'stream1', ephemeralStreamId: 'stream1-eph', adminStreamId: 'stream1-3', + name: 'Channel', type: 'public', createdAt: 1000, joinedAt: 1000, + members: ['0x1'], storageDays: 30, ...extra + }); + const saved = () => secureStorage.setChannels.mock.calls.at(-1)[0][0]; + + beforeEach(() => { + secureStorage.isStorageUnlocked.mockReturnValue(true); + }); + + it('stamps only the fields the save changes', async () => { + secureStorage.getChannels.mockReturnValue([stored()]); + channelManager.loadChannels(); + channelManager.channels.get('stream1').name = 'Renamed'; + + await channelManager.saveChannels(); + + expect(Object.keys(saved().fieldTs)).toEqual(['name']); + expect(channelManager.channels.get('stream1').fieldTs).toEqual(saved().fieldTs); + }); + + it('stamps a list changed in place', async () => { + secureStorage.getChannels.mockReturnValue([stored()]); + channelManager.loadChannels(); + channelManager.channels.get('stream1').members.push('0x2'); + + await channelManager.saveChannels(); + + expect(Object.keys(saved().fieldTs)).toEqual(['members']); + }); + + it('stamps nothing when a save changes nothing, and keeps the stamps it had', async () => { + secureStorage.getChannels.mockReturnValue([stored({ fieldTs: { name: 5000 } })]); + channelManager.loadChannels(); + + await channelManager.saveChannels(); + + expect(saved().fieldTs).toEqual({ name: 5000 }); + }); + + it('does not stamp what a sync pull brought in', async () => { + secureStorage.getChannels.mockReturnValue([stored()]); + channelManager.loadChannels(); + secureStorage.getChannels.mockReturnValue([stored({ name: 'From another device', fieldTs: { name: 5000 } })]); + channelManager.reloadChannelsFromSync(); + + await channelManager.saveChannels(); + + expect(saved().name).toBe('From another device'); + expect(saved().fieldTs).toEqual({ name: 5000 }); + }); + + it('does not stamp a record created here', async () => { + secureStorage.getChannels.mockReturnValue([]); + channelManager.loadChannels(); + channelManager.channels.set('stream1', stored()); + + await channelManager.saveChannels(); + + expect(saved().fieldTs).toBeUndefined(); + }); + }); + describe('getCachedDeletePermission()', () => { beforeEach(() => { authManager.getAddress.mockReturnValue('0xmyaddress'); @@ -2353,18 +2421,20 @@ describe('ChannelManager', () => { // ==================== member_update via handleControlMessage ==================== describe('handleControlMessage() - member_update', () => { - it('should update channel members on member_update', async () => { + // No client publishes it and nothing checks who sent it: a list taken + // from the wire would be saved, stamped and synced to every device. + it('leaves the members as they are and saves nothing', async () => { const channel = { members: ['0x1'], reactions: {} }; channelManager.channels.set('stream1', channel); secureStorage.isStorageUnlocked.mockReturnValue(true); - secureStorage.setChannels.mockResolvedValue(undefined); await channelManager.handleControlMessage('stream1', { type: 'member_update', members: ['0x1', '0x2', '0x3'] }); - expect(channel.members).toEqual(['0x1', '0x2', '0x3']); + expect(channel.members).toEqual(['0x1']); + expect(secureStorage.setChannels).not.toHaveBeenCalled(); }); }); diff --git a/tests/unit/syncMerge.vectors.test.js b/tests/unit/syncMerge.vectors.test.js index 4ebf90d..3208843 100644 --- a/tests/unit/syncMerge.vectors.test.js +++ b/tests/unit/syncMerge.vectors.test.js @@ -27,6 +27,14 @@ describe('sync merge parity vectors', () => { }); } + for (const v of vectors.channels) { + it(v.what, () => { + const merged = mergeState(v.base, v.incoming); + expect(merged.channels).toEqual(v.expected.channels); + expect(merged.channelsLeftAt).toEqual(v.expected.channelsLeftAt); + }); + } + for (const v of vectors.sent) { it(v.what, () => { const merged = mergeState(v.base, v.incoming); diff --git a/tests/vectors/gen_sync_merge_vectors.mjs b/tests/vectors/gen_sync_merge_vectors.mjs index 4a3b453..d1b72aa 100644 --- a/tests/vectors/gen_sync_merge_vectors.mjs +++ b/tests/vectors/gen_sync_merge_vectors.mjs @@ -8,7 +8,8 @@ // // The vectors fix the slice outcome of one merge step (base = this device, // incoming = a remote snapshot), the stamping of unstamped values before a -// state leaves the device, and how sent DMs carry deletions and edits. +// state leaves the device, how sent DMs carry deletions and edits, and how two +// copies of a channel record merge. import { mergeState, stampedSliceTs } from '../../src/js/syncMerge.js'; const SLICES = ['blockedPeers', 'dmLeftAt', 'trustedContacts', 'username', 'graphApiKey']; @@ -40,6 +41,20 @@ const merge = (what, base, incoming) => ({ const DM = '0x00000000000000000000000000000000000000a1/Pombo-DM-1'; const text = (id, timestamp, extra = {}) => ({ id, type: 'text', text: `text of ${id}`, timestamp, ...extra }); +const CH = '0x00000000000000000000000000000000000000a1/c0ffee-1'; +const record = (extra = {}) => ({ + messageStreamId: CH, name: 'Channel', type: 'public', createdAt: 1000, joinedAt: 1000, + storageDays: 30, accessSnapshot: [], ...extra +}); + +// Channel records: each field comes from the copy that stamped it later; the +// join-time rule only decides what no stamp does, and whether the channel is +// joined at all. +const channel = (what, base, incoming) => { + const merged = mergeState(base, incoming); + return { what, base, incoming, expected: { channels: merged.channels, channelsLeftAt: merged.channelsLeftAt } }; +}; + // Sent DMs: a deletion on any device removes the message on every device, // and the latest edit wins. const sent = (what, base, incoming) => { @@ -106,5 +121,37 @@ console.log(JSON.stringify({ sent('a deletion from another device wins over a later edit held here', { sentMessages: { [DM]: [text('m1', 1000, { text: 'later', _edited: true, _editedAt: 9000 })] } }, { sentMessages: { [DM]: [] }, sentDeletedAt: { [DM]: { m1: 5000 } } }) + ], + channels: [ + channel('a stamped field held here beats an old unstamped copy arriving', + { channels: [record({ name: 'Renamed', fieldTs: { name: 5000 } })] }, + { channels: [record()] }), + channel('a stamped field arriving beats the old unstamped copy held here', + { channels: [record()] }, + { channels: [record({ name: 'Renamed', fieldTs: { name: 5000 } })] }), + channel('two devices changing different fields keep both changes', + { channels: [record({ name: 'Renamed', fieldTs: { name: 5000 } })] }, + { channels: [record({ accessSnapshot: ['0x00000000000000000000000000000000000000b1'], fieldTs: { accessSnapshot: 6000 } })] }), + channel('the same two changes merged in the other order give the same record', + { channels: [record({ accessSnapshot: ['0x00000000000000000000000000000000000000b1'], fieldTs: { accessSnapshot: 6000 } })] }, + { channels: [record({ name: 'Renamed', fieldTs: { name: 5000 } })] }), + channel('a client that drops the stamps neither reverts a stamped field nor erases the stamps', + { channels: [record({ storageDays: 90, fieldTs: { storageDays: 5000 } })] }, + { channels: [record({ storageDays: 30 })] }), + channel('an older stamp arriving does not undo a newer one held here', + { channels: [record({ name: 'Newer', fieldTs: { name: 6000 } })] }, + { channels: [record({ name: 'Older', fieldTs: { name: 5000 } })] }), + channel('equal stamps fall back to the join-time rule, which gives the tie to the incoming copy', + { channels: [record({ name: 'Here', fieldTs: { name: 5000 } })] }, + { channels: [record({ name: 'Arriving', fieldTs: { name: 5000 } })] }), + channel('a field only one copy carries is kept, whichever copy wins the others', + { channels: [record({ keysStreamId: `${CH.slice(0, -2)}-4`, storageProvider: 'streamr' })] }, + { channels: [record({ inboxStreamId: null, name: 'Renamed', fieldTs: { name: 5000 } })] }), + channel('a leave newer than the join removes the channel whatever its stamps say', + { channels: [record({ name: 'Renamed', fieldTs: { name: 9000 } })] }, + { channels: [], channelsLeftAt: { [CH]: 5000 } }), + channel('a join newer than the leave keeps the channel and retires the leave', + { channels: [record({ joinedAt: 7000 })] }, + { channels: [], channelsLeftAt: { [CH]: 5000 } }) ] }, null, 2));