package eventbus

import "github.com/segmentio/kafka-go"

// MessagesAfterCommittedPrefixes keeps every fetched message that is not
// covered by the highest committed offset for its topic partition. This is
// used when a batch partially succeeds: later poison or valid messages must
// remain in memory until the failed offset ahead of them is durably handled.
func MessagesAfterCommittedPrefixes(messages []kafka.Message, committed []kafka.Message) []kafka.Message {
	type partitionKey struct {
		topic     string
		partition int
	}
	highest := make(map[partitionKey]int64, len(committed))
	for _, message := range committed {
		key := partitionKey{topic: message.Topic, partition: message.Partition}
		if offset, ok := highest[key]; !ok || message.Offset > offset {
			highest[key] = message.Offset
		}
	}
	remaining := make([]kafka.Message, 0, len(messages))
	for _, message := range messages {
		key := partitionKey{topic: message.Topic, partition: message.Partition}
		if offset, ok := highest[key]; ok && message.Offset <= offset {
			continue
		}
		remaining = append(remaining, message)
	}
	return remaining
}