Servise gelenen belge güncellemeleri message Document { string Url = 1; // Belgenin URL'si, benzersiz tanımlayıcısı uint64 PubDate = 2; // belgenin bildirilen yayınlama zamanı uint64 FetchTime = 3; // bu belge güncellemesinin alınma zamanı, sürüm tanımlayıcısı olarak düşünülebilir. Çift (Url, FetchTime) benzersizdir. string Text = 4; // belgenin metni uint64 FirstFetchTime = 5; // başlangıçta yok, doldurulmalı } Belgeler herhangi bir sırayla gelebilir (güncellemelerin yapıldığı sırayla değil), ayrıca mesaj tekrarları da olabilir. Çıkışta, aynı mesajlar oluşturmak gerekiyor, ancak aşağıdaki kurallara göre düzeltilmiş alanlarla (aşağıda belirtilenler, aynı Url alanına sahip belge grubu için geçerlidir): Text ve FetchTime alanları, şu ana kadar alınan en büyük FetchTime'a sahip belgeyle aynı olmalı. PubDate alanı, en küçük FetchTime'a sahip mesajla aynı olmalı. FirstFetchTime alanı, en küçük FetchTime değerine eşit olmalı. Yani, her zaman, FetchTime'a göre sıralanmış ilk versiyondan alınan PubDate ve FirstFetchTime, ve en son versiyondan alınan Text kullanılır. Kodda bu arayüz şu şekilde uygulanabilir: type Processor interface { Process(doc *Document) (*Document, error) } Bu kod, mesajları bir kuyruktan (Kafka veya benzeri) okuyan ve sonucu da kuyrukta yazan bir serviste çalışacaktır. Eğer Process null dönerse, kuyrukta hiçbir şey yazılmaz.

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