tencent-baidu-tracking/src/main/java/com/adx/tencent/kuaishou/service/KuaishouColdWorker.java

package com.adx.tencent.kuaishou.service;

import com.adx.tencent.kuaishou.model.KuaishouBidRecord;
import com.adx.tencent.kuaishou.model.KuaishouTrackingRecord;
import com.adx.tencent.kuaishou.store.KuaishouColdStore;
import com.adx.tencent.kuaishou.store.KuaishouHotStore;
import com.adx.tencent.storage.model.QueuedEvent;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

public class KuaishouColdWorker {

    private final KuaishouHotStore hotStore;
    private final KuaishouColdStore coldStore;
    private final String group;
    private final String consumer;
    private final int batch;
    private final Duration pendingIdle;
    private final long streamMaxLen;
    private final ObjectMapper objectMapper;

    public KuaishouColdWorker(KuaishouHotStore hotStore, KuaishouColdStore coldStore,
                           String group, String consumer, int batch,
                           Duration pendingIdle, long streamMaxLen,
                           ObjectMapper objectMapper) {
        this.hotStore = hotStore;
        this.coldStore = coldStore;
        this.group = group;
        this.consumer = consumer;
        this.batch = batch <= 0 ? 100 : batch;
        this.pendingIdle = pendingIdle == null || pendingIdle.isZero() ? Duration.ofMinutes(2) : pendingIdle;
        this.streamMaxLen = streamMaxLen;
        this.objectMapper = objectMapper;
    }

    public int runOnce() throws Exception {
        hotStore.ensureGroup(group);
        List<QueuedEvent> events = hotStore.claimStale(group, consumer, pendingIdle, batch);
        if (events.isEmpty()) events = hotStore.read(group, consumer, batch);
        if (events.isEmpty()) return 0;

        List<String> acked = new ArrayList<>();
        Exception failed = null;
        for (QueuedEvent event : events) {
            try {
                handleEvent(event);
            } catch (Exception e) {
                failed = e;
                break;
            }
            acked.add(event.getId());
        }
        if (failed != null) throw failed;
        hotStore.ack(group, acked);
        hotStore.delete(acked);
        if (streamMaxLen > 0) hotStore.trim(streamMaxLen);
        return events.size();
    }

    private void handleEvent(QueuedEvent event) throws Exception {
        switch (event.getType()) {
            case "bid" -> coldStore.saveBid(objectMapper.readValue(event.getPayload(), KuaishouBidRecord.class));
            case "tracking" -> coldStore.saveTracking(objectMapper.readValue(event.getPayload(), KuaishouTrackingRecord.class));
            default -> throw new IllegalArgumentException("unknown kuaishou event type: " + event.getType());
        }
    }
}