Փաստաթղթերի թարմացումները գալիս են ծառայություն message Document { string Url = 1; // Փաստաթղթի URL- ը, նրա եզակի նույնականացուցիչ uint64 PubDate = 2; // հայտարարած հրապարակման ժամանակը uint64 FetchTime = 3; // այս փաստաթղթի թարմացման ստացման ժամանակը, կարող է դիտվել որպես տարբերակի նույնականացուցիչ։ Զույգ (Url, FetchTime) եզակի է։ string Text = 4; // փաստաթղթի տեքստը uint64 FirstFetchTime = 5; // սկզբում բացակայում է, պետք է լրացվի } Փաստաթղթերը կարող են գալ ցանկացած կարգով (ոչ այն կարգով, որով դրանք թարմացվել են), և հնարավոր է կրկնօրինակ հաղորդագրություններ: Ելքում անհրաժեշտ է ձևավորել նույն հաղորդագրությունները, բայց ուղղված հետևյալ կանոններին (այստեղ ամեն ինչ վերաբերում է նույն Url դաշտով խմբավորված փաստաթղթերին): Text և FetchTime դաշտերը պետք է լինեն այնպիսին, ինչպիսին եղել է մինչ այժմ ստացված ամենամեծ FetchTime-ով փաստաթղթում: PubDate դաշտը պետք է լինի այնպիսին, ինչպիսին եղել է ամենափոքր FetchTime-ով հաղորդագրությունում: FirstFetchTime դաշտը պետք է լինի ամենափոքր FetchTime-ով արժեքը: Այնպես որ, յուրաքանչյուր պահի մենք վերցնում ենք PubDate և FirstFetchTime- ը առաջին տարբերակից, որը մինչ այժմ ստացվել է (եթե դրանք դասավորենք FetchTime-ով), և Text- ը վերջինից: Կոդում այս ինտերֆեյսը կարող է իրականացվել այսպես: type Processor interface { Process(doc *Document) (*Document, error) } Այս կոդը կաշխատի ծառայությունում, որը կարդում է հաղորդագրություններ հերթից (Kafka կամ նման), և նաև գրում է արդյունքը հերթին: Եթե Process վերադարձնում է Null, ոչինչ չի գրանցվում հերթում:

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