Οι ενημερώσεις εγγράφων φτάνουν στην υπηρεσία 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 и возвращать скорректированные документы согласно правилам.