Actualizările documentelor sosesc la serviciu message Document { string Url = 1; // URL-ul documentului, identificatorul său unic uint64 PubDate = 2; // timpul de publicare declarat al documentului uint64 FetchTime = 3; // timpul de primire a acestei actualizări a documentului, poate fi considerat ca un identificator de versiune. Perechea (Url, FetchTime) este unică. string Text = 4; // textul documentului uint64 FirstFetchTime = 5; // inițial absent, trebuie completat } Documentele pot ajunge în orice ordine (nu în ordinea în care au fost actualizate), și pot exista mesaje duplicate. Este necesar să se formeze în ieșire mesaje identice, dar cu câmpuri corectate conform următoarelor reguli (tot ce este mai jos se aplică pentru un grup de documente cu același câmp Url): Câmpul Text și FetchTime trebuie să fie cele ale documentului cu cel mai mare FetchTime primit până acum. Câmpul PubDate trebuie să fie cel al mesajului cu cel mai mic FetchTime. Câmpul FirstFetchTime trebuie să fie egal cu valoarea minimă a FetchTime. Cu alte cuvinte, în fiecare moment, luăm PubDate și FirstFetchTime din prima versiune primită până acum (dacă le sortăm după FetchTime), și Text din ultima. Interfața în cod poate fi implementată astfel: type Processor interface { Process(doc *Document) (*Document, error) } Acest cod va funcționa într-un serviciu care citește mesaje dintr-o coadă (Kafka sau similar), și scrie și rezultatul în coadă. Dacă Process returnează Null, nimic nu se scrie în coadă.

Для реализации интерфейса Processor в Go нужно хранить состояние по каждому Url, чтобы на каждый входящий документ обновлять агрегированное состояние и возвращать скорректированное сообщение.

Основная идея:

• Для каждого Url хранить:
  • минимальный FetchTime (для PubDate и FirstFetchTime)
  • максимальный FetchTime (для Text и FetchTime)
  • PubDate, Text, FirstFetchTime, соответствующие этим минимальному и максимальному FetchTime

При получении нового документа:

• Если для Url нет записи — создать с данными из документа
• Иначе обновить минимальный и максимальный FetchTime и соответствующие поля при необходимости
• Вернуть агрегированный документ с нужными полями

Пример реализации:

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 {
        // первый документ для данного 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

    // Обновляем минимальный FetchTime и PubDate, FirstFetchTime
    if doc.FetchTime < st.minFetchTime {
        st.minFetchTime = doc.FetchTime
        st.pubDate = doc.PubDate
        st.firstFetch = doc.FetchTime
        updated = true
    }

    // Обновляем максимальный FetchTime и Text
    if doc.FetchTime > st.maxFetchTime {
        st.maxFetchTime = doc.FetchTime
        st.text = doc.Text
        updated = true
    }

    // Если пришло сообщение с FetchTime, который уже есть, но текст или PubDate отличаются,
    // можно игнорировать, так как пара (Url, FetchTime) уникальна, либо обновлять по логике.

    if !updated {
        // Данные не изменились, можно вернуть nil, чтобы не писать в очередь
        return nil, nil
    }

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

Таким образом, сервис будет поддерживать актуальное агрегированное состояние для каждого Url и возвращать скорректированные документы согласно правилам.