Skip to content

fix(gift): stop streak gifts being dropped and double counted #160

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lulajax wants to merge 1 commit into
base: master
Choose a base branch
from
+202 −32
Open

fix(gift): stop streak gifts being dropped and double counted #160

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
125 changes: 97 additions & 28 deletions ...t/src/main/java/io/github/jwdeveloper/tiktok/mappers/handlers/TikTokGiftEventHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,16 +34,48 @@
import lombok.SneakyThrows;

import java.util.*;
import java.util.function.LongSupplier;

public class TikTokGiftEventHandler {
private final Map<Long, WebcastGiftMessage> giftsMessages;
/**
* A streak that has not been finished within this window is closed by its last active frame.
* TikTok does not guarantee a finishing frame (sendType 0) for every streak, and without this
* fallback such a gift would never raise onGift at all.
* <p>
* Measured from the last frame of the streak rather than its start, so an ongoing streak keeps
* extending it. Deliberately generous: streaks can run into the thousands and a lull between
* frames must not close one early, or the gift is split in half and the real finishing frame
* is then discarded as a duplicate.
*/
private static final long DEFAULT_COMBO_TIMEOUT_MS = 300_000;

/**
* TikTok repeats the finishing frame of a streak under a fresh msgId within milliseconds.
* Finishing frames arriving inside this window after a streak was closed are ignored.
*/
private static final long FINALIZED_RETENTION_MS = 60_000;

private final Map<String, ComboState> activeCombos;
private final Map<String, Long> finalizedCombos;
private final TikTokRoomInfo tikTokRoomInfo;
private final GiftsManager giftsManager;
private final long comboTimeoutMillis;
private final LongSupplier clock;

public TikTokGiftEventHandler(GiftsManager giftsManager, TikTokRoomInfo tikTokRoomInfo) {
giftsMessages = new HashMap<>();
this(giftsManager, tikTokRoomInfo, DEFAULT_COMBO_TIMEOUT_MS, System::currentTimeMillis);
}

public TikTokGiftEventHandler(GiftsManager giftsManager,
TikTokRoomInfo tikTokRoomInfo,
long comboTimeoutMillis,
LongSupplier clock) {
this.activeCombos = new HashMap<>();
this.finalizedCombos = new HashMap<>();
this.tikTokRoomInfo = tikTokRoomInfo;
this.giftsManager = giftsManager;
this.comboTimeoutMillis = comboTimeoutMillis;
this.clock = clock;
}

@SneakyThrows
Expand All @@ -54,43 +86,80 @@ public MappingResult handleGifts(byte[] msg, String name, LiveMapperHelper helpe
}

public List<TikTokEvent> handleGift(WebcastGiftMessage currentMessage) {
var now = clock.getAsLong();
var events = new ArrayList<>(flushTimedOutCombos(now));

//If gift is not streakable just return onGift event
if (currentMessage.getGift().getType() != 1) {
var comboEvent = getGiftComboEvent(currentMessage, GiftComboStateType.Finished);
var giftEvent = getGiftEvent(currentMessage);
return List.of(comboEvent, giftEvent);
events.add(getGiftComboEvent(currentMessage, GiftComboStateType.Finished));
events.add(getGiftEvent(currentMessage));
return events;
}

var userId = currentMessage.getUser().getId();
var key = comboKey(currentMessage);
var currentType = GiftComboStateType.fromNumber(currentMessage.getSendType());
var previousMessage = giftsMessages.get(userId);

if (previousMessage == null) {
if (currentType == GiftComboStateType.Finished) {
return List.of(getGiftEvent(currentMessage));
} else {
giftsMessages.put(userId, currentMessage);
return List.of(getGiftComboEvent(currentMessage, GiftComboStateType.Begin));
}
}

var previousType = GiftComboStateType.fromNumber(previousMessage.getSendType());
if (currentType == GiftComboStateType.Active &&
previousType == GiftComboStateType.Active) {
giftsMessages.put(userId, currentMessage);
return List.of(getGiftComboEvent(currentMessage, GiftComboStateType.Active));
if (currentType == GiftComboStateType.Active) {
var previous = activeCombos.put(key, new ComboState(currentMessage, now));
events.add(getGiftComboEvent(currentMessage,
previous == null ? GiftComboStateType.Begin : GiftComboStateType.Active));
return events;
}

//TikTok may repeat the finishing frame of a streak under a fresh msgId. Without this guard
//every repeat raises another onGift and the gift gets counted twice.
var finalizedUntil = finalizedCombos.get(key);
if (finalizedUntil != null && finalizedUntil > now)
return events;

var previous = activeCombos.remove(key);
finalizedCombos.put(key, now + FINALIZED_RETENTION_MS);
if (previous != null)
events.add(getGiftComboEvent(currentMessage, GiftComboStateType.Finished));
events.add(getGiftEvent(currentMessage));
return events;
}

if (currentType == GiftComboStateType.Finished &&
previousType == GiftComboStateType.Active) {
giftsMessages.clear();
return List.of(
getGiftComboEvent(currentMessage, GiftComboStateType.Finished),
getGiftEvent(currentMessage));
/**
* Closes streaks that timed out waiting for their finishing frame. Runs lazily on every
* incoming gift, so a streak left open at the very end of a live may still be missed.
*/
private List<TikTokEvent> flushTimedOutCombos(long now) {
finalizedCombos.values().removeIf(expiresAt -> expiresAt <= now);
if (activeCombos.isEmpty())
return List.of();

var events = new ArrayList<TikTokEvent>();
var iterator = activeCombos.entrySet().iterator();
while (iterator.hasNext()) {
var entry = iterator.next();
var state = entry.getValue();
if (now - state.updatedAt() < comboTimeoutMillis)
continue;

iterator.remove();
finalizedCombos.put(entry.getKey(), now + FINALIZED_RETENTION_MS);
events.add(getGiftComboEvent(state.message(), GiftComboStateType.Finished));
events.add(getGiftEvent(state.message()));
}
return events;
}

/**
* Identifies a single streak. Keying on the user alone lets concurrent streaks of the same
* user overwrite each other, and made one finishing frame wipe the state of every other user.
*/
private String comboKey(WebcastGiftMessage message) {
var groupId = message.getGroupId() != 0
? Long.toString(message.getGroupId())
: message.getOrderId();
if (groupId == null || groupId.isEmpty())
groupId = Long.toString(message.getCommon().getMsgId());

return message.getUser().getId() + ":" + message.getGiftId() + ":" + groupId;
}

return List.of();
private record ComboState(WebcastGiftMessage message, long updatedAt) {
}


Expand Down
109 changes: 105 additions & 4 deletions ...rc/test/java/io/github/jwdeveloper/tiktok/handlers/events/TikTokGiftEventHandlerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,25 +34,30 @@
import io.github.jwdeveloper.tiktok.messages.data.User;
import io.github.jwdeveloper.tiktok.messages.webcast.WebcastGiftMessage;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;

import java.util.List;
import java.util.concurrent.atomic.AtomicLong;


@TestInstance(TestInstance.Lifecycle.PER_CLASS)
class TikTokGiftEventHandlerTest {

public static TikTokGiftEventHandler handler;
private static final long COMBO_TIMEOUT_MS = 30_000;

@BeforeAll
public TikTokGiftEventHandler handler;
private AtomicLong clock;

@BeforeEach
public void before() {
var manager = new TikTokGiftsManager(List.of());
var info = new TikTokRoomInfo();
info.setHost(new io.github.jwdeveloper.tiktok.data.models.users.User(123L, "test", new Picture("")));
manager.attachGift(new Gift(123, "example", 123, "image.webp"));
handler = new TikTokGiftEventHandler(manager, info);
clock = new AtomicLong(1_000L);
handler = new TikTokGiftEventHandler(manager, info, COMBO_TIMEOUT_MS, clock::get);
}

@Test
Expand Down Expand Up @@ -103,12 +108,107 @@ void shouldHandleStrike() {
}


@Test
void shouldKeepConcurrentStreaksOfSameUserApart() {
var streakA = getGiftMessage("example-new-name", 123, "image-new.png", 1, 1, true, 111);
var streakB = getGiftMessage("example-new-name", 123, "image-new.png", 1, 1, true, 222);

Assertions.assertEquals(GiftComboStateType.Begin, comboStateOf(handler.handleGift(streakA)));
Assertions.assertEquals(GiftComboStateType.Begin, comboStateOf(handler.handleGift(streakB)));

var finishA = getGiftMessage("example-new-name", 123, "image-new.png", 0, 1, true, 111);
var finishB = getGiftMessage("example-new-name", 123, "image-new.png", 0, 1, true, 222);

//Both streaks must finish on their own, keying on the user alone used to drop one of them
Assertions.assertEquals(1, countGiftEvents(handler.handleGift(finishA)));
Assertions.assertEquals(1, countGiftEvents(handler.handleGift(finishB)));
}

@Test
void shouldNotDropStreakOfOtherUserWhenOneFinishes() {
var userOneActive = getGiftMessage("example-new-name", 123, "image-new.png", 1, 1, true, 111);
var userTwoActive = getGiftMessage("example-new-name", 123, "image-new.png", 1, 2, true, 222);
handler.handleGift(userOneActive);
handler.handleGift(userTwoActive);

//Finishing user two used to clear() the whole map and reset user one back to Begin
handler.handleGift(getGiftMessage("example-new-name", 123, "image-new.png", 0, 2, true, 222));

var next = handler.handleGift(getGiftMessage("example-new-name", 123, "image-new.png", 1, 1, true, 111));
Assertions.assertEquals(GiftComboStateType.Active, comboStateOf(next));
}

@Test
void shouldIgnoreRepeatedFinishFrame() {
handler.handleGift(getGiftMessage("example-new-name", 123, "image-new.png", 1, 1, true, 111));

var first = handler.handleGift(getGiftMessage("example-new-name", 123, "image-new.png", 0, 1, true, 111));
Assertions.assertEquals(1, countGiftEvents(first));

//TikTok resends the finishing frame under a fresh msgId; it must not raise onGift again
clock.addAndGet(150);
var repeated = handler.handleGift(getGiftMessage("example-new-name", 123, "image-new.png", 0, 1, true, 111));
Assertions.assertEquals(0, countGiftEvents(repeated));
}

@Test
void shouldNotCutLongRunningStreak() {
//A streak of 1000 keeps sending frames; the timeout is measured from the last one, so it
//must never fire mid-streak and split the gift in two
for (var i = 1; i <= 1000; i++) {
clock.addAndGet(COMBO_TIMEOUT_MS / 2);
var frame = handler.handleGift(
getGiftMessage("example-new-name", 123, "image-new.png", 1, 1, true, 111));
Assertions.assertEquals(0, countGiftEvents(frame), "streak was cut at frame " + i);
}

//Only the real finishing frame closes it, carrying the full repeat count
var finish = handler.handleGift(getGiftMessage("example-new-name", 123, "image-new.png", 0, 1, true, 111));
Assertions.assertEquals(1, countGiftEvents(finish));
}

@Test
void shouldCloseStreakThatNeverReceivedFinishFrame() {
handler.handleGift(getGiftMessage("example-new-name", 123, "image-new.png", 4, 1, true, 111));

//Any later gift drives the lazy sweep; the abandoned streak is closed by its last frame
clock.addAndGet(COMBO_TIMEOUT_MS + 1);
var unrelated = getGiftMessage("example-new-name", 123, "image-new.png", 0, 9, false, 999);
var result = handler.handleGift(unrelated);

//One for the timed out streak, one for the gift that triggered the sweep
Assertions.assertEquals(2, countGiftEvents(result));
}

private GiftComboStateType comboStateOf(List<io.github.jwdeveloper.tiktok.data.events.common.TikTokEvent> events) {
return events.stream()
.filter(TikTokGiftComboEvent.class::isInstance)
.map(TikTokGiftComboEvent.class::cast)
.findFirst()
.orElseThrow()
.getComboState();
}

private long countGiftEvents(List<io.github.jwdeveloper.tiktok.data.events.common.TikTokEvent> events) {
return events.stream().filter(e -> !(e instanceof TikTokGiftComboEvent)).count();
}

public WebcastGiftMessage getGiftMessage(String giftName,
int giftId,
String giftImage,
int sendType,
int userId,
boolean streakable) {
return getGiftMessage(giftName, giftId, giftImage, sendType, userId, streakable, 0);
}

public WebcastGiftMessage getGiftMessage(String giftName,
int giftId,
String giftImage,
int sendType,
int userId,
boolean streakable,
long groupId) {
var builder = WebcastGiftMessage.newBuilder();
var giftBuilder = io.github.jwdeveloper.tiktok.messages.data.Gift.newBuilder();
var userBuilder = User.newBuilder();
Expand All @@ -123,6 +223,7 @@ public WebcastGiftMessage getGiftMessage(String giftName,
builder.setGiftId(giftId);
builder.setUser(userBuilder);
builder.setSendType(sendType);
builder.setGroupId(groupId);
builder.setGift(giftBuilder);
return builder.build();
}
Expand Down