Aktualisierungen von Dokumenten werden an den Dienst gesendet message Document { string Url = 1; // URL des Dokuments, seine eindeutige Kennung uint64 PubDate = 2; // deklarierte Veröffentlichungszeit des Dokuments uint64 FetchTime = 3; // Empfangszeit dieser Dokumentenaktualisierung, kann als Versionskennung betrachtet werden. Das Paar (Url, FetchTime) ist eindeutig. string Text = 4; // Text des Dokuments uint64 FirstFetchTime = 5; // initially absent, must be filled } Dokumente können in beliebiger Reihenfolge eintreffen (nicht in der Reihenfolge, in der sie aktualisiert wurden), und es können doppelte Nachrichten vorhanden sein. Es ist notwendig, in der Ausgabe gleiche Nachrichten zu erstellen, aber mit korrigierten Feldern nach den folgenden Regeln (alles unten gilt für eine Gruppe von Dokumenten mit demselben Feld Url): Das Feld Text und FetchTime sollten so sein wie bei dem Dokument mit der größten FetchTime, die bisher empfangen wurde. Das Feld PubDate sollte so sein wie bei der Nachricht mit der kleinsten FetchTime. Das Feld FirstFetchTime sollte den minimalen FetchTime-Wert haben. Das heißt, zu jedem Zeitpunkt nehmen wir PubDate und FirstFetchTime vom ersten Version, die bisher empfangen wurde (wenn man sie nach FetchTime sortiert), und Text von der letzten. Die Schnittstelle im Code kann so implementiert werden: type Processor interface { Process(doc *Document) (*Document, error) } Dieser Code wird in einem Dienst funktionieren, der Nachrichten aus einer Warteschlange liest (Kafka oder ähnlich), und auch das Ergebnis in die Warteschlange schreibt. Wenn Process Null zurückgibt, wird nichts in die Warteschlange geschrieben.

Для реализации интерфейса 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 и возвращать скорректированные документы согласно правилам.