Merge pull request #2454 from yougotwill/sogs_poller_conflict_resolver
Opengroup Poller Reaction Conflict Resolution + Fixespull/2460/head
Audric Ackermann
3 years ago
committed by
GitHub
commit
25f54a13ff
@ -0,0 +1,135 @@
|
|||||||
|
/**
|
||||||
|
* This is strictly use to resolve conflicts between local state and the opengroup poll updates
|
||||||
|
* Currently only supports message reactions 26/08/2022
|
||||||
|
*/
|
||||||
|
|
||||||
|
import { filter, findIndex, remove } from 'lodash';
|
||||||
|
import { Reactions } from '../../../../util/reactions';
|
||||||
|
import { OpenGroupReactionMessageV4 } from '../opengroupV2/OpenGroupServerPoller';
|
||||||
|
|
||||||
|
export enum ChangeType {
|
||||||
|
REACTIONS = 0,
|
||||||
|
}
|
||||||
|
|
||||||
|
type ReactionAction = 'ADD' | 'REMOVE' | 'CLEAR';
|
||||||
|
|
||||||
|
type ReactionChange = {
|
||||||
|
messageId: number; // will be serverId of the reacted message
|
||||||
|
emoji: string;
|
||||||
|
action: ReactionAction;
|
||||||
|
};
|
||||||
|
|
||||||
|
export type SogsV3Mutation = {
|
||||||
|
seqno: number | null; // null until mutating API request returns
|
||||||
|
server: string; // serverUrl
|
||||||
|
room: string; // roomId
|
||||||
|
changeType: ChangeType;
|
||||||
|
metadata: ReactionChange; // For now we only support message reactions
|
||||||
|
};
|
||||||
|
|
||||||
|
// we don't want to export this, we want to export functions that manipulate it
|
||||||
|
const sogsMutationCache: Array<SogsV3Mutation> = [];
|
||||||
|
|
||||||
|
// for testing purposes only
|
||||||
|
export function getMutationCache() {
|
||||||
|
return sogsMutationCache;
|
||||||
|
}
|
||||||
|
|
||||||
|
function verifyEntry(entry: SogsV3Mutation): boolean {
|
||||||
|
return Boolean(
|
||||||
|
entry.server &&
|
||||||
|
entry.room &&
|
||||||
|
entry.changeType === ChangeType.REACTIONS &&
|
||||||
|
entry.metadata.messageId &&
|
||||||
|
entry.metadata.emoji &&
|
||||||
|
(entry.metadata.action === 'ADD' ||
|
||||||
|
entry.metadata.action === 'REMOVE' ||
|
||||||
|
entry.metadata.action === 'CLEAR')
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
export function addToMutationCache(entry: SogsV3Mutation) {
|
||||||
|
if (!verifyEntry(entry)) {
|
||||||
|
window.log.error('SOGS Mutation Cache: Entry verification on add failed!', entry);
|
||||||
|
} else {
|
||||||
|
sogsMutationCache.push(entry);
|
||||||
|
window.log.info('SOGS Mutation Cache: Entry added!', entry);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
export function updateMutationCache(entry: SogsV3Mutation, seqno: number) {
|
||||||
|
if (!verifyEntry(entry)) {
|
||||||
|
window.log.error('SOGS Mutation Cache: Entry verification on update failed!', entry);
|
||||||
|
} else {
|
||||||
|
const entryIndex = findIndex(sogsMutationCache, entry);
|
||||||
|
if (entryIndex >= 0) {
|
||||||
|
sogsMutationCache[entryIndex].seqno = seqno;
|
||||||
|
window.log.info('SOGS Mutation Cache: Entry updated!', sogsMutationCache[entryIndex]);
|
||||||
|
} else {
|
||||||
|
window.log.error('SOGS Mutation Cache: Updated failed! Cannot find entry', entry);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// return is for testing purposes only
|
||||||
|
export async function processMessagesUsingCache(
|
||||||
|
server: string,
|
||||||
|
room: string,
|
||||||
|
message: OpenGroupReactionMessageV4
|
||||||
|
): Promise<OpenGroupReactionMessageV4> {
|
||||||
|
const updatedReactions = message.reactions;
|
||||||
|
|
||||||
|
const roomMatches: Array<SogsV3Mutation> = filter(sogsMutationCache, { server, room });
|
||||||
|
for (let i = 0; i < roomMatches.length; i++) {
|
||||||
|
const matchSeqno = roomMatches[i].seqno;
|
||||||
|
if (message.seqno && matchSeqno && matchSeqno <= message.seqno) {
|
||||||
|
const removedEntry = roomMatches.splice(i, 1)[0];
|
||||||
|
window.log.info('SOGS Mutation Cache: Entry ignored and removed!', removedEntry);
|
||||||
|
remove(sogsMutationCache, removedEntry);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
for (const reaction of Object.keys(message.reactions)) {
|
||||||
|
const reactionMatches = filter(roomMatches, {
|
||||||
|
server,
|
||||||
|
room,
|
||||||
|
changeType: ChangeType.REACTIONS,
|
||||||
|
metadata: {
|
||||||
|
messageId: message.id,
|
||||||
|
emoji: reaction,
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
for (const reactionMatch of reactionMatches) {
|
||||||
|
switch (reactionMatch.metadata.action) {
|
||||||
|
case 'ADD':
|
||||||
|
updatedReactions[reaction].you = true;
|
||||||
|
updatedReactions[reaction].count += 1;
|
||||||
|
window.log.info(
|
||||||
|
'SOGS Mutation Cache: Added our reaction based on the cache',
|
||||||
|
updatedReactions[reaction]
|
||||||
|
);
|
||||||
|
break;
|
||||||
|
case 'REMOVE':
|
||||||
|
updatedReactions[reaction].you = false;
|
||||||
|
updatedReactions[reaction].count -= 1;
|
||||||
|
window.log.info(
|
||||||
|
'SOGS Mutation Cache: Removed our reaction based on the cache',
|
||||||
|
updatedReactions[reaction]
|
||||||
|
);
|
||||||
|
break;
|
||||||
|
default:
|
||||||
|
window.log.warn(
|
||||||
|
'SOGS Mutation Cache: Unsupported metadata action in OpenGroupMessageV4',
|
||||||
|
reactionMatch
|
||||||
|
);
|
||||||
|
}
|
||||||
|
const removedEntry = remove(sogsMutationCache, reactionMatch);
|
||||||
|
window.log.info('SOGS Mutation Cache: Entry removed!', removedEntry);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
message.reactions = updatedReactions;
|
||||||
|
await Reactions.handleOpenGroupMessageReactions(message.reactions, message.id);
|
||||||
|
return message;
|
||||||
|
}
|
||||||
@ -0,0 +1,365 @@
|
|||||||
|
import { expect } from 'chai';
|
||||||
|
import Sinon from 'sinon';
|
||||||
|
import {
|
||||||
|
addToMutationCache,
|
||||||
|
ChangeType,
|
||||||
|
getMutationCache,
|
||||||
|
processMessagesUsingCache,
|
||||||
|
SogsV3Mutation,
|
||||||
|
updateMutationCache,
|
||||||
|
} from '../../../../session/apis/open_group_api/sogsv3/sogsV3MutationCache';
|
||||||
|
import { TestUtils } from '../../../test-utils';
|
||||||
|
import { Reactions } from '../../../../util/reactions';
|
||||||
|
import {
|
||||||
|
OpenGroupMessageV4,
|
||||||
|
OpenGroupReactionMessageV4,
|
||||||
|
} from '../../../../session/apis/open_group_api/opengroupV2/OpenGroupServerPoller';
|
||||||
|
// tslint:disable: chai-vague-errors
|
||||||
|
|
||||||
|
describe('mutationCache', () => {
|
||||||
|
TestUtils.stubWindowLog();
|
||||||
|
|
||||||
|
const roomInfos = TestUtils.generateOpenGroupV2RoomInfos();
|
||||||
|
const originalMessage = TestUtils.generateOpenGroupMessageV2WithServerId(111);
|
||||||
|
const originalMessage2 = TestUtils.generateOpenGroupMessageV2WithServerId(112);
|
||||||
|
const reactor1 = TestUtils.generateFakePubKey().key;
|
||||||
|
const reactor2 = TestUtils.generateFakePubKey().key;
|
||||||
|
|
||||||
|
beforeEach(() => {
|
||||||
|
// stubs
|
||||||
|
Sinon.stub(Reactions, 'handleOpenGroupMessageReactions').resolves();
|
||||||
|
});
|
||||||
|
|
||||||
|
afterEach(Sinon.restore);
|
||||||
|
|
||||||
|
describe('add entry to cache', () => {
|
||||||
|
it('add entry to cache that is valid', () => {
|
||||||
|
const entry: SogsV3Mutation = {
|
||||||
|
server: roomInfos.serverUrl,
|
||||||
|
room: roomInfos.roomId,
|
||||||
|
changeType: ChangeType.REACTIONS,
|
||||||
|
seqno: null,
|
||||||
|
metadata: {
|
||||||
|
messageId: originalMessage.serverId,
|
||||||
|
emoji: '😄',
|
||||||
|
action: 'ADD',
|
||||||
|
},
|
||||||
|
};
|
||||||
|
addToMutationCache(entry);
|
||||||
|
const cache = getMutationCache();
|
||||||
|
expect(cache, 'should not empty').to.not.equal([]);
|
||||||
|
expect(cache.length, 'should have one entry').to.be.equal(1);
|
||||||
|
expect(cache[0], 'the entry should match the input').to.be.deep.equal(entry);
|
||||||
|
});
|
||||||
|
it('add entry to cache that is invalid and fail', () => {
|
||||||
|
const entry: SogsV3Mutation = {
|
||||||
|
server: '', // this is invalid
|
||||||
|
room: roomInfos.roomId,
|
||||||
|
changeType: ChangeType.REACTIONS,
|
||||||
|
seqno: 100,
|
||||||
|
metadata: {
|
||||||
|
messageId: originalMessage.serverId,
|
||||||
|
emoji: '😄',
|
||||||
|
action: 'ADD',
|
||||||
|
},
|
||||||
|
};
|
||||||
|
addToMutationCache(entry);
|
||||||
|
const cache = getMutationCache();
|
||||||
|
expect(cache, 'should not empty').to.not.equal([]);
|
||||||
|
expect(cache.length, 'should have one entry').to.be.equal(1);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
describe('update entry in cache', () => {
|
||||||
|
it('update entry in cache with a valid source entry', () => {
|
||||||
|
const entry: SogsV3Mutation = {
|
||||||
|
server: roomInfos.serverUrl,
|
||||||
|
room: roomInfos.roomId,
|
||||||
|
changeType: ChangeType.REACTIONS,
|
||||||
|
seqno: null, // mutation before we have received a response
|
||||||
|
metadata: {
|
||||||
|
messageId: originalMessage.serverId,
|
||||||
|
emoji: '😄',
|
||||||
|
action: 'ADD',
|
||||||
|
},
|
||||||
|
};
|
||||||
|
const messageResponse = TestUtils.generateFakeIncomingOpenGroupMessageV4({
|
||||||
|
id: originalMessage.serverId,
|
||||||
|
seqno: 200,
|
||||||
|
reactions: {
|
||||||
|
'😄': {
|
||||||
|
index: 0,
|
||||||
|
count: 1,
|
||||||
|
you: false,
|
||||||
|
reactors: [reactor1],
|
||||||
|
},
|
||||||
|
'❤️': {
|
||||||
|
index: 1,
|
||||||
|
count: 2,
|
||||||
|
you: true,
|
||||||
|
reactors: [originalMessage.sender, reactor1],
|
||||||
|
},
|
||||||
|
'😈': {
|
||||||
|
index: 2,
|
||||||
|
count: 2,
|
||||||
|
you: true,
|
||||||
|
reactors: [originalMessage.sender, reactor2],
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}) as OpenGroupMessageV4;
|
||||||
|
updateMutationCache(entry, (messageResponse as OpenGroupMessageV4).seqno);
|
||||||
|
const cache = getMutationCache();
|
||||||
|
expect(cache, 'should not empty').to.not.equal([]);
|
||||||
|
expect(cache.length, 'should have one entry').to.be.equal(1);
|
||||||
|
expect(
|
||||||
|
cache[0].seqno,
|
||||||
|
'should have an entry with a matching seqno to the message response'
|
||||||
|
).to.be.equal(messageResponse.seqno);
|
||||||
|
});
|
||||||
|
it('update entry in cache with an invalid source entry', () => {
|
||||||
|
const messageResponse = TestUtils.generateFakeIncomingOpenGroupMessageV4({
|
||||||
|
id: originalMessage.serverId,
|
||||||
|
seqno: 200,
|
||||||
|
reactions: {
|
||||||
|
'😄': {
|
||||||
|
index: 0,
|
||||||
|
count: 1,
|
||||||
|
you: false,
|
||||||
|
reactors: [reactor1],
|
||||||
|
},
|
||||||
|
'❤️': {
|
||||||
|
index: 1,
|
||||||
|
count: 2,
|
||||||
|
you: true,
|
||||||
|
reactors: [originalMessage.sender, reactor1],
|
||||||
|
},
|
||||||
|
'😈': {
|
||||||
|
index: 2,
|
||||||
|
count: 2,
|
||||||
|
you: true,
|
||||||
|
reactors: [originalMessage.sender, reactor2],
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}) as OpenGroupMessageV4;
|
||||||
|
const entry: SogsV3Mutation = {
|
||||||
|
server: '',
|
||||||
|
room: roomInfos.roomId,
|
||||||
|
changeType: ChangeType.REACTIONS,
|
||||||
|
seqno: 100,
|
||||||
|
metadata: {
|
||||||
|
messageId: originalMessage.serverId,
|
||||||
|
emoji: '😄',
|
||||||
|
action: 'ADD',
|
||||||
|
},
|
||||||
|
};
|
||||||
|
updateMutationCache(entry, (messageResponse as OpenGroupMessageV4).seqno);
|
||||||
|
const cache = getMutationCache();
|
||||||
|
expect(cache, 'should not empty').to.not.equal([]);
|
||||||
|
expect(cache.length, 'should have one entry').to.be.equal(1);
|
||||||
|
expect(
|
||||||
|
cache[0].seqno,
|
||||||
|
'should have an entry with a matching seqno to the message response'
|
||||||
|
).to.be.equal(messageResponse.seqno);
|
||||||
|
});
|
||||||
|
it('update entry in cache with a valid source entry but its not stored in the cache', () => {
|
||||||
|
const messageResponse = TestUtils.generateFakeIncomingOpenGroupMessageV4({
|
||||||
|
id: originalMessage.serverId,
|
||||||
|
seqno: 200,
|
||||||
|
reactions: {
|
||||||
|
'😄': {
|
||||||
|
index: 0,
|
||||||
|
count: 1,
|
||||||
|
you: false,
|
||||||
|
reactors: [reactor1],
|
||||||
|
},
|
||||||
|
'❤️': {
|
||||||
|
index: 1,
|
||||||
|
count: 2,
|
||||||
|
you: true,
|
||||||
|
reactors: [originalMessage.sender, reactor1],
|
||||||
|
},
|
||||||
|
'😈': {
|
||||||
|
index: 2,
|
||||||
|
count: 2,
|
||||||
|
you: true,
|
||||||
|
reactors: [originalMessage.sender, reactor2],
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}) as OpenGroupMessageV4;
|
||||||
|
const entry: SogsV3Mutation = {
|
||||||
|
server: roomInfos.serverUrl,
|
||||||
|
room: roomInfos.roomId,
|
||||||
|
changeType: ChangeType.REACTIONS,
|
||||||
|
seqno: 400,
|
||||||
|
metadata: {
|
||||||
|
messageId: originalMessage.serverId,
|
||||||
|
emoji: '😄',
|
||||||
|
action: 'ADD',
|
||||||
|
},
|
||||||
|
};
|
||||||
|
updateMutationCache(entry, (messageResponse as OpenGroupMessageV4).seqno);
|
||||||
|
const cache = getMutationCache();
|
||||||
|
expect(cache, 'should not empty').to.not.equal([]);
|
||||||
|
expect(cache.length, 'should have one entry').to.be.equal(1);
|
||||||
|
expect(
|
||||||
|
cache[0].seqno,
|
||||||
|
'should have an entry with a matching seqno to the message response'
|
||||||
|
).to.be.equal(messageResponse.seqno);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
describe('process opengroup messages using the cache', () => {
|
||||||
|
it('processing a message with valid serverUrl, roomId and message should return the same message response', async () => {
|
||||||
|
const messageResponse = TestUtils.generateFakeIncomingOpenGroupMessageV4({
|
||||||
|
id: originalMessage.serverId,
|
||||||
|
seqno: 200,
|
||||||
|
reactions: {
|
||||||
|
'😄': {
|
||||||
|
index: 0,
|
||||||
|
count: 1,
|
||||||
|
you: false,
|
||||||
|
reactors: [reactor1],
|
||||||
|
},
|
||||||
|
'❤️': {
|
||||||
|
index: 1,
|
||||||
|
count: 2,
|
||||||
|
you: true,
|
||||||
|
reactors: [originalMessage.sender, reactor1],
|
||||||
|
},
|
||||||
|
'😈': {
|
||||||
|
index: 2,
|
||||||
|
count: 2,
|
||||||
|
you: true,
|
||||||
|
reactors: [originalMessage.sender, reactor2],
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}) as OpenGroupMessageV4;
|
||||||
|
const message = await processMessagesUsingCache(
|
||||||
|
roomInfos.serverUrl,
|
||||||
|
roomInfos.roomId,
|
||||||
|
messageResponse
|
||||||
|
);
|
||||||
|
const cache = getMutationCache();
|
||||||
|
expect(cache, 'cache should be empty').to.be.empty;
|
||||||
|
expect(message, 'message response should match').to.be.deep.equal(messageResponse);
|
||||||
|
});
|
||||||
|
it('processing a message with valid serverUrl, roomId and message (from SOGS < 1.3.4) should return the same message response', async () => {
|
||||||
|
const messageResponse = TestUtils.generateFakeIncomingOpenGroupMessageV4({
|
||||||
|
id: originalMessage2.serverId,
|
||||||
|
// in version less than 1.3.4 there is no a seqno set
|
||||||
|
reactions: {
|
||||||
|
'🤣': {
|
||||||
|
index: 0,
|
||||||
|
count: 3,
|
||||||
|
you: true,
|
||||||
|
reactors: [reactor1, reactor2, originalMessage2.sender],
|
||||||
|
},
|
||||||
|
'😈': {
|
||||||
|
index: 0,
|
||||||
|
count: 1,
|
||||||
|
you: false,
|
||||||
|
reactors: [reactor2],
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}) as OpenGroupReactionMessageV4;
|
||||||
|
const message = await processMessagesUsingCache(
|
||||||
|
roomInfos.serverUrl,
|
||||||
|
roomInfos.roomId,
|
||||||
|
messageResponse
|
||||||
|
);
|
||||||
|
const cache = getMutationCache();
|
||||||
|
expect(cache, 'cache should be empty').to.be.empty;
|
||||||
|
expect(message, 'message response should match').to.be.deep.equal(messageResponse);
|
||||||
|
});
|
||||||
|
it('processing a message with valid entries in the cache should calculate the optimistic state if there is no message seqo or the cached entry seqno is larger than the message seqno', async () => {
|
||||||
|
const messageResponse = TestUtils.generateFakeIncomingOpenGroupMessageV4({
|
||||||
|
id: originalMessage.serverId,
|
||||||
|
seqno: 200,
|
||||||
|
reactions: {
|
||||||
|
'😄': {
|
||||||
|
index: 0,
|
||||||
|
count: 1,
|
||||||
|
you: false,
|
||||||
|
reactors: [reactor1],
|
||||||
|
},
|
||||||
|
'❤️': {
|
||||||
|
index: 1,
|
||||||
|
count: 2,
|
||||||
|
you: true,
|
||||||
|
reactors: [originalMessage.sender, reactor1],
|
||||||
|
},
|
||||||
|
'😈': {
|
||||||
|
index: 2,
|
||||||
|
count: 2,
|
||||||
|
you: true,
|
||||||
|
reactors: [originalMessage.sender, reactor2],
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}) as OpenGroupMessageV4;
|
||||||
|
const entry: SogsV3Mutation = {
|
||||||
|
server: roomInfos.serverUrl,
|
||||||
|
room: roomInfos.roomId,
|
||||||
|
changeType: ChangeType.REACTIONS,
|
||||||
|
seqno: 100, // less than response messageResponse seqno should be ignored
|
||||||
|
metadata: {
|
||||||
|
messageId: originalMessage.serverId,
|
||||||
|
emoji: '❤️',
|
||||||
|
action: 'ADD',
|
||||||
|
},
|
||||||
|
};
|
||||||
|
const entry2: SogsV3Mutation = {
|
||||||
|
server: roomInfos.serverUrl,
|
||||||
|
room: roomInfos.roomId,
|
||||||
|
changeType: ChangeType.REACTIONS,
|
||||||
|
seqno: 300, // greater than response messageResponse seqno should be procesed
|
||||||
|
metadata: {
|
||||||
|
messageId: originalMessage.serverId,
|
||||||
|
emoji: '😄',
|
||||||
|
action: 'ADD',
|
||||||
|
},
|
||||||
|
};
|
||||||
|
const entry3: SogsV3Mutation = {
|
||||||
|
server: roomInfos.serverUrl,
|
||||||
|
room: roomInfos.roomId,
|
||||||
|
changeType: ChangeType.REACTIONS,
|
||||||
|
seqno: 301, //// greater than response messageResponse seqno should be procesed
|
||||||
|
metadata: {
|
||||||
|
messageId: originalMessage.serverId,
|
||||||
|
emoji: '😈',
|
||||||
|
action: 'REMOVE',
|
||||||
|
},
|
||||||
|
};
|
||||||
|
addToMutationCache(entry);
|
||||||
|
addToMutationCache(entry2);
|
||||||
|
addToMutationCache(entry3);
|
||||||
|
|
||||||
|
const message = await processMessagesUsingCache(
|
||||||
|
roomInfos.serverUrl,
|
||||||
|
roomInfos.roomId,
|
||||||
|
messageResponse
|
||||||
|
);
|
||||||
|
const cache = getMutationCache();
|
||||||
|
expect(cache, 'cache should be empty').to.be.empty;
|
||||||
|
expect(
|
||||||
|
message.reactions['❤️'].count,
|
||||||
|
'message response reaction count for ❤️ should be unchanged with 2'
|
||||||
|
).to.equal(2);
|
||||||
|
expect(
|
||||||
|
message.reactions['😄'].count,
|
||||||
|
'message response reaction count for 😄 should be 2'
|
||||||
|
).to.equal(2);
|
||||||
|
expect(
|
||||||
|
message.reactions['😄'].you,
|
||||||
|
'message response reaction for 😄 should have you = true'
|
||||||
|
).to.equal(true);
|
||||||
|
expect(
|
||||||
|
message.reactions['😈'].count,
|
||||||
|
'message response reaction count for 😈 should be 1'
|
||||||
|
).to.equal(1);
|
||||||
|
expect(
|
||||||
|
message.reactions['😈'].you,
|
||||||
|
'message response reaction for 😈 should have you = false'
|
||||||
|
).to.equal(false);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
});
|
||||||
Loading…
Reference in New Issue