Documents are received by the service message Document { string Url = 1; // Document URL, its unique identifier uint64 PubDate = 2; // Document publication time uint64 FetchTime = 3; // Time of receiving this document update, can be considered as version ID. Pair (Url, FetchTime) is unique. string Text = 4; // Document text uint64 FirstFetchTime = 5; // Initially absent, needs to be filled } Documents can arrive in any order (not in the order they were updated), and duplicates of individual messages are also possible. It is necessary to form the same messages at the output, but with corrected individual fields according to the following rules (for a group of documents with the same Url field): Text and FetchTime fields should be the latest received at the moment (the one with the highest FetchTime) PubDate should be the one from the message with the smallest FetchTime FirstFetchTime should be the minimum FetchTime value In other words, at each moment, we take PubDate and FirstFetchTime from the earliest version (sorted by FetchTime), and Text from the latest. The interface in code can be implemented as follows: type Processor interface { Process(doc *Document) (*Document, error) } This code will work in a service reading input messages from a message queue (Kafka or similar), and writing the result back to the queue. If Process returns Null - nothing is written to the queue.

To implement the Processor interface in Go, you need to store the state for each URL to update the aggregated state with each incoming document and return the corrected message.

Main idea:

• For each URL, store:
  • the minimum FetchTime (for PubDate and FirstFetchTime)
  • the maximum FetchTime (for Text and FetchTime)
  • PubDate, Text, FirstFetchTime corresponding to these minimum and maximum FetchTime

When a new document is received:

• If there is no record for the URL — create one with data from the document
• Otherwise, update the minimum and maximum FetchTime and the corresponding fields if necessary
• Return the aggregated document with the required fields

Example implementation:

import (
    "errors"
    "sync"
)

type Document struct {
    Url            string
    PubDate        uint64
    FetchTime      uint64
    Text           string
    FirstFetchTime uint64
}

type Processor interface {
    Process(doc *Document) (*Document, error)
}

type docState struct {
    minFetchTime uint64
    maxFetchTime uint64
    pubDate      uint64
    text         string
    firstFetch   uint64
}

type processorImpl struct {
    mu    sync.Mutex
    state map[string]*docState
}

func NewProcessor() Processor {
    return &processorImpl{
        state: make(map[string]*docState),
    }
}

func (p *processorImpl) Process(doc *Document) (*Document, error) {
    if doc == nil {
        return nil, errors.New("input document is nil")
    }

    p.mu.Lock()
    defer p.mu.Unlock()

    st, ok := p.state[doc.Url]
    if !ok {
        // first document for this URL
        p.state[doc.Url] = &docState{
            minFetchTime: doc.FetchTime,
            maxFetchTime: doc.FetchTime,
            pubDate:      doc.PubDate,
            text:         doc.Text,
            firstFetch:   doc.FetchTime,
        }
        return &Document{
            Url:            doc.Url,
            PubDate:        doc.PubDate,
            FetchTime:      doc.FetchTime,
            Text:           doc.Text,
            FirstFetchTime: doc.FetchTime,
        }, nil
    }

    updated := false

    // Update minimum FetchTime and PubDate, FirstFetchTime
    if doc.FetchTime < st.minFetchTime {
        st.minFetchTime = doc.FetchTime
        st.pubDate = doc.PubDate
        st.firstFetch = doc.FetchTime
        updated = true
    }

    // Update maximum FetchTime and Text
    if doc.FetchTime > st.maxFetchTime {
        st.maxFetchTime = doc.FetchTime
        st.text = doc.Text
        updated = true
    }

    // If a message with an existing FetchTime arrives, but the text or PubDate differ,
    // it can be ignored, as the pair (Url, FetchTime) is unique, or updated according to logic.

    if !updated {
        // Data has not changed, can return nil to avoid queueing
        return nil, nil
    }

    return &Document{
        Url:            doc.Url,
        PubDate:        st.pubDate,
        FetchTime:      st.maxFetchTime,
        Text:           st.text,
        FirstFetchTime: st.firstFetch,
    }, nil
}

Thus, the service will maintain the current aggregated state for each URL and return corrected documents according to the rules.