Sincronizzazione tra un'app online e una offline con CloudEvents: l'implementazione in Go
Questo è il primo di una serie di 7 articoli in cui costruiamo, passo per passo, lo stesso meccanismo di sincronizzazione tra un'applicazione online (un server centrale, sempre raggiungibile) e un'applicazione offline (un client che passa la maggior parte del tempo disconnesso e si collega solo per sincronizzarsi). Il design è identico in tutti gli articoli: cambia solo lo stack usato per realizzarlo. Nei prossimi vedremo la stessa architettura in Node.js, Laravel, Python, PHP puro, Java Spring Boot e C#. Qui partiamo da Go.
Il problema
Un client che è offline per la maggior parte del tempo (un'app desktop, una postazione di sala, un dispositivo sul campo) non può permettersi di dipendere da una connessione sempre attiva per funzionare. Deve continuare a operare normalmente sui dati locali e sincronizzarsi con il server centrale solo quando la rete torna disponibile, senza bloccarsi e senza perdere nulla nel frattempo.
Il pattern che risolve questo problema si chiama outbox/inbox: ogni cambiamento locale non viene inviato subito al server, ma accodato in una tabella (o file) locale, l'outbox. Un processo separato prova periodicamente a svuotare l'outbox verso il server e, allo stesso tempo, scarica ciò che è successo altrove (l'inbox) applicandolo al dominio locale. Se la rete manca, il tentativo fallisce silenziosamente e riprova al giro successivo: l'app continua a funzionare, la sync è solo un processo asincrono in background.
Il formato dell'evento: CloudEvents
Perché due applicazioni scritte in stack diversi possano scambiarsi eventi senza inventarsi ogni volta un formato nuovo, conviene appoggiarsi a uno standard esistente: CloudEvents 1.0, la specifica della CNCF per descrivere un evento in modo indipendente dal linguaggio e dal protocollo di trasporto. In JSON un evento CloudEvents ha questo aspetto:
{
"specversion": "1.0",
"id": "d7cb8234ce639d51408afe0eeb679c5c",
"type": "com.gabrieleromanato.offlineapp.record.created",
"source": "urn:client:offline-client-01",
"time": "2026-09-18T16:15:30Z",
"datacontenttype": "application/json",
"data": {
"recordId": 42,
"note": "Creato mentre offline"
}
}id è univoco e garantisce l'idempotenza: se lo stesso evento arriva due volte per un retry di rete, il server lo scarta alla seconda ricezione. type segue una convenzione reverse-DNS per evitare collisioni tra applicazioni diverse, e source identifica chi ha generato l'evento.
L'architettura di sincronizzazione
Il server online espone tre operazioni: push, con cui il client invia il batch di eventi accumulati nell'outbox; pull, con cui il client scarica gli eventi generati altrove da quando si è sincronizzato l'ultima volta; ack, con cui il client conferma fino a dove ha applicato gli eventi ricevuti, così il server avanza il cursore associato a quel client. Ogni client ha un proprio cursore lato server, un numero di sequenza che indica il punto della sincronizzazione a cui è arrivato.
Per tenere l'esempio eseguibile senza dipendenze esterne, la persistenza qui è un semplice file JSON protetto da un mutex, sia lato server che lato client: in un progetto reale lo sostituiresti con un database (PostgreSQL lato server, SQLite lato client), ma la logica di sincronizzazione resterebbe la stessa.
L'app online: il server di sync
Il pacchetto net/http della libreria standard, dalla versione 1.22 di Go, supporta pattern con metodo HTTP direttamente nel ServeMux ("POST /api/sync/push"), quindi non serve nessun router esterno. Il event store è isolato in un file a parte:
// store.go — event store append-only, persistito su file JSON.
// Nessuna dipendenza esterna: per un servizio reale si userebbe un database
// (PostgreSQL, SQLite...), ma per l'esempio basta la libreria standard.
package main
import (
"encoding/json"
"os"
"sync"
)
type storedEvent struct {
Seq int64 `json:"seq"`
Ev CloudEvent `json:"event"`
}
type EventStore struct {
mu sync.Mutex
eventsPath string
cursorsPath string
events []storedEvent
seenIDs map[string]bool
nextSeq int64
clientCursor map[string]int64
}
func NewEventStore(eventsPath, cursorsPath string) *EventStore {
s := &EventStore{
eventsPath: eventsPath,
cursorsPath: cursorsPath,
seenIDs: map[string]bool{},
clientCursor: map[string]int64{},
nextSeq: 1,
}
s.load()
return s
}
func (s *EventStore) load() {
if data, err := os.ReadFile(s.eventsPath); err == nil {
_ = json.Unmarshal(data, &s.events)
for _, se := range s.events {
s.seenIDs[se.Ev.ID] = true
if se.Seq >= s.nextSeq {
s.nextSeq = se.Seq + 1
}
}
}
if data, err := os.ReadFile(s.cursorsPath); err == nil {
_ = json.Unmarshal(data, &s.clientCursor)
}
}
func (s *EventStore) persistEvents() error {
data, err := json.MarshalIndent(s.events, "", " ")
if err != nil {
return err
}
return os.WriteFile(s.eventsPath, data, 0o644)
}
func (s *EventStore) persistCursors() error {
data, err := json.MarshalIndent(s.clientCursor, "", " ")
if err != nil {
return err
}
return os.WriteFile(s.cursorsPath, data, 0o644)
}
// InsertIfNew inserisce l'evento solo se il suo id non è già stato visto (idempotenza)
func (s *EventStore) InsertIfNew(ev CloudEvent) (inserted bool, err error) {
s.mu.Lock()
defer s.mu.Unlock()
if s.seenIDs[ev.ID] {
return false, nil
}
se := storedEvent{Seq: s.nextSeq, Ev: ev}
s.events = append(s.events, se)
s.seenIDs[ev.ID] = true
s.nextSeq++
if err := s.persistEvents(); err != nil {
return false, err
}
return true, nil
}
func (s *EventStore) Since(seq int64, limit int) []storedEvent {
s.mu.Lock()
defer s.mu.Unlock()
out := make([]storedEvent, 0, limit)
for _, se := range s.events {
if se.Seq > seq {
out = append(out, se)
if len(out) >= limit {
break
}
}
}
return out
}
func (s *EventStore) Cursor(clientID string) int64 {
s.mu.Lock()
defer s.mu.Unlock()
return s.clientCursor[clientID]
}
func (s *EventStore) SetCursor(clientID string, seq int64) error {
s.mu.Lock()
defer s.mu.Unlock()
s.clientCursor[clientID] = seq
return s.persistCursors()
}E il server vero e proprio, con i tre endpoint e l'autenticazione via API key:
// main.go — app online: server di sincronizzazione basato su CloudEvents 1.0
package main
import (
"encoding/json"
"log"
"net/http"
"os"
"time"
)
type CloudEvent struct {
SpecVersion string `json:"specversion"`
ID string `json:"id"`
Type string `json:"type"`
Source string `json:"source"`
Time string `json:"time"`
DataContentType string `json:"datacontenttype,omitempty"`
Data json.RawMessage `json:"data"`
}
func (e CloudEvent) isValid() bool {
return e.SpecVersion == "1.0" && e.ID != "" && e.Type != "" && e.Source != "" && e.Time != ""
}
// Autenticazione minimale via API key statica (in produzione: JWT o mTLS)
func requireAPIKey(next http.HandlerFunc) http.HandlerFunc {
apiKey := os.Getenv("API_KEY")
if apiKey == "" {
apiKey = "dev-secret"
}
return func(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("X-Api-Key") != apiKey {
http.Error(w, `{"error":"unauthorized"}`, http.StatusUnauthorized)
return
}
next(w, r)
}
}
type pushRequest struct {
ClientID string `json:"clientId"`
Events []CloudEvent `json:"events"`
}
type pushResult struct {
ID string `json:"id"`
Inserted bool `json:"inserted,omitempty"`
Reason string `json:"reason,omitempty"`
}
func handlePush(store *EventStore) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
var req pushRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil || req.ClientID == "" {
http.Error(w, `{"error":"clientId ed events[] sono obbligatori"}`, http.StatusBadRequest)
return
}
accepted := []pushResult{}
rejected := []pushResult{}
for _, ev := range req.Events {
if !ev.isValid() {
rejected = append(rejected, pushResult{ID: ev.ID, Reason: "evento non conforme a CloudEvents 1.0"})
continue
}
inserted, err := store.InsertIfNew(ev)
if err != nil {
http.Error(w, `{"error":"errore interno"}`, http.StatusInternalServerError)
return
}
// inserted=false => id già visto in precedenza, scartato per idempotenza
accepted = append(accepted, pushResult{ID: ev.ID, Inserted: inserted})
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]any{
"accepted": accepted,
"rejected": rejected,
"serverTime": time.Now().UTC().Format(time.RFC3339),
})
}
}
func handlePull(store *EventStore) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
clientID := r.URL.Query().Get("clientId")
if clientID == "" {
http.Error(w, `{"error":"clientId obbligatorio"}`, http.StatusBadRequest)
return
}
since := store.Cursor(clientID)
events := store.Since(since, 200)
cursor := since
out := make([]CloudEvent, 0, len(events))
for _, se := range events {
out = append(out, se.Ev)
cursor = se.Seq
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]any{"events": out, "cursor": cursor})
}
}
type ackRequest struct {
ClientID string `json:"clientId"`
Cursor int64 `json:"cursor"`
}
func handleAck(store *EventStore) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
var req ackRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil || req.ClientID == "" {
http.Error(w, `{"error":"clientId e cursor sono obbligatori"}`, http.StatusBadRequest)
return
}
if err := store.SetCursor(req.ClientID, req.Cursor); err != nil {
http.Error(w, `{"error":"errore interno"}`, http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]bool{"ok": true})
}
}
func main() {
store := NewEventStore("events.json", "cursors.json")
mux := http.NewServeMux()
mux.HandleFunc("POST /api/sync/push", requireAPIKey(handlePush(store)))
mux.HandleFunc("GET /api/sync/pull", requireAPIKey(handlePull(store)))
mux.HandleFunc("POST /api/sync/ack", requireAPIKey(handleAck(store)))
port := os.Getenv("PORT")
if port == "" {
port = "3000"
}
log.Printf("Online sync server in ascolto su :%s", port)
log.Fatal(http.ListenAndServe(":"+port, mux))
}L'app offline: outbox locale e servizio di sync
Lato client la struttura è simmetrica: un outbox per gli eventi in uscita e un registro degli eventi già applicati, per restare idempotenti anche qui.
// store.go — outbox locale e registro degli eventi applicati, su file JSON.
package main
import (
"encoding/json"
"os"
"sync"
)
type outboxItem struct {
Ev CloudEvent `json:"event"`
Sent bool `json:"sent"`
}
type LocalStore struct {
mu sync.Mutex
outboxPath string
appliedPath string
outbox []outboxItem
applied map[string]bool
}
func NewLocalStore(outboxPath, appliedPath string) *LocalStore {
s := &LocalStore{
outboxPath: outboxPath,
appliedPath: appliedPath,
applied: map[string]bool{},
}
s.load()
return s
}
func (s *LocalStore) load() {
if data, err := os.ReadFile(s.outboxPath); err == nil {
_ = json.Unmarshal(data, &s.outbox)
}
if data, err := os.ReadFile(s.appliedPath); err == nil {
var ids []string
if json.Unmarshal(data, &ids) == nil {
for _, id := range ids {
s.applied[id] = true
}
}
}
}
func (s *LocalStore) persistOutbox() error {
data, err := json.MarshalIndent(s.outbox, "", " ")
if err != nil {
return err
}
return os.WriteFile(s.outboxPath, data, 0o644)
}
func (s *LocalStore) persistApplied() error {
ids := make([]string, 0, len(s.applied))
for id := range s.applied {
ids = append(ids, id)
}
data, err := json.MarshalIndent(ids, "", " ")
if err != nil {
return err
}
return os.WriteFile(s.appliedPath, data, 0o644)
}
// Enqueue viene chiamato dal dominio offline ogni volta che avviene un cambiamento locale
func (s *LocalStore) Enqueue(ev CloudEvent) error {
s.mu.Lock()
defer s.mu.Unlock()
s.outbox = append(s.outbox, outboxItem{Ev: ev})
return s.persistOutbox()
}
func (s *LocalStore) Unsent(limit int) []CloudEvent {
s.mu.Lock()
defer s.mu.Unlock()
out := make([]CloudEvent, 0, limit)
for _, item := range s.outbox {
if !item.Sent {
out = append(out, item.Ev)
if len(out) >= limit {
break
}
}
}
return out
}
func (s *LocalStore) MarkSent(ids []string) error {
s.mu.Lock()
defer s.mu.Unlock()
idSet := make(map[string]bool, len(ids))
for _, id := range ids {
idSet[id] = true
}
for i := range s.outbox {
if idSet[s.outbox[i].Ev.ID] {
s.outbox[i].Sent = true
}
}
return s.persistOutbox()
}
// ApplyIncoming applica un evento ricevuto dal server al dominio locale;
// idempotente grazie al controllo su applied[ev.ID].
func (s *LocalStore) ApplyIncoming(ev CloudEvent) error {
s.mu.Lock()
defer s.mu.Unlock()
if s.applied[ev.ID] {
return nil // già applicato: evita doppie applicazioni
}
s.applied[ev.ID] = true
// Qui va la logica reale: deserializzare ev.Data in base a ev.Type
// e aggiornare l'entità corrispondente nel modello di dominio offline.
return s.persistApplied()
}E il servizio che, a intervalli regolari, prova a sincronizzarsi:
// main.go — app offline: outbox locale e servizio di sync periodico
package main
import (
"bytes"
"crypto/rand"
"encoding/hex"
"encoding/json"
"log"
"net/http"
"time"
)
type CloudEvent struct {
SpecVersion string `json:"specversion"`
ID string `json:"id"`
Type string `json:"type"`
Source string `json:"source"`
Time string `json:"time"`
DataContentType string `json:"datacontenttype,omitempty"`
Data json.RawMessage `json:"data"`
}
func newEventID() string {
b := make([]byte, 16)
_, _ = rand.Read(b)
return hex.EncodeToString(b)
}
type syncClient struct {
baseURL string
apiKey string
clientID string
store *LocalStore
http *http.Client
}
type pullResponse struct {
Events []CloudEvent `json:"events"`
Cursor int64 `json:"cursor"`
}
func (c *syncClient) trySync() error {
// 1) PUSH: invia gli eventi locali accumulati mentre si era offline
pending := c.store.Unsent(200)
if len(pending) > 0 {
body, _ := json.Marshal(map[string]any{"clientId": c.clientID, "events": pending})
req, _ := http.NewRequest(http.MethodPost, c.baseURL+"/api/sync/push", bytes.NewReader(body))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("X-Api-Key", c.apiKey)
resp, err := c.http.Do(req)
if err != nil {
return err // rete assente: normale per un client offline-first
}
resp.Body.Close()
ids := make([]string, len(pending))
for i, ev := range pending {
ids[i] = ev.ID
}
if err := c.store.MarkSent(ids); err != nil {
return err
}
log.Printf("Inviati %d eventi al server", len(pending))
}
// 2) PULL: scarica gli eventi generati altrove dopo l'ultima sync
req, _ := http.NewRequest(http.MethodGet, c.baseURL+"/api/sync/pull?clientId="+c.clientID, nil)
req.Header.Set("X-Api-Key", c.apiKey)
resp, err := c.http.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
var payload pullResponse
if err := json.NewDecoder(resp.Body).Decode(&payload); err != nil {
return err
}
for _, ev := range payload.Events {
if err := c.store.ApplyIncoming(ev); err != nil {
return err
}
}
// 3) ACK: conferma al server fino a dove si è applicato
if len(payload.Events) > 0 {
ackBody, _ := json.Marshal(map[string]any{"clientId": c.clientID, "cursor": payload.Cursor})
ackReq, _ := http.NewRequest(http.MethodPost, c.baseURL+"/api/sync/ack", bytes.NewReader(ackBody))
ackReq.Header.Set("Content-Type", "application/json")
ackReq.Header.Set("X-Api-Key", c.apiKey)
ackResp, err := c.http.Do(ackReq)
if err != nil {
return err
}
ackResp.Body.Close()
log.Printf("Applicati %d eventi ricevuti dal server", len(payload.Events))
}
return nil
}
func main() {
store := NewLocalStore("outbox.json", "applied.json")
client := &syncClient{
baseURL: "http://localhost:3000",
apiKey: "dev-secret",
clientID: "offline-client-01",
store: store,
http: &http.Client{Timeout: 10 * time.Second},
}
// Esempio: simula un cambiamento avvenuto nel dominio offline.
// Nella tua app reale questa chiamata va fatta subito dopo ogni
// scrittura locale rilevante.
_ = store.Enqueue(CloudEvent{
SpecVersion: "1.0",
ID: newEventID(),
Type: "com.gabrieleromanato.offlineapp.record.created",
Source: "urn:client:" + client.clientID,
Time: time.Now().UTC().Format(time.RFC3339),
DataContentType: "application/json",
Data: json.RawMessage(`{"recordId":42,"note":"Creato mentre offline"}`),
})
ticker := time.NewTicker(30 * time.Second)
defer ticker.Stop()
for {
if err := client.trySync(); err != nil {
log.Printf("Sync non riuscita (rete assente?): %v", err)
}
<-ticker.C
}
}Provarlo end-to-end
In due terminali separati:
cd online-go && go mod init syncserver && go run .
# in un altro terminale
cd offline-go && go mod init syncclient && go run .Al primo avvio il client accoda un evento di esempio, lo invia al server con push e poi lo scarica di nuovo con pull (perché non c'è ancora un filtro sulla provenienza), applicandolo idempotentemente al proprio registro locale. Spegni il server per qualche secondo: il client continua a girare, i tentativi falliscono silenziosamente nei log e riprendono non appena il server torna su, senza intervento manuale.
Cosa manca per la produzione
Questo esempio è pensato per essere chiaro e interamente eseguibile senza dipendenze esterne, non per andare in produzione così com'è. Prima di usarlo per davvero servono almeno: un database reale al posto del file JSON (il mutex in memoria non scala oltre un singolo processo); un client che non si riapplichi da solo gli eventi che ha generato, filtrando per source; autenticazione più solida (JWT con refresh o mTLS) al posto della API key statica; retry con backoff esponenziale invece del semplice intervallo fisso; e una strategia esplicita di conflict resolution se due client modificano lo stesso record mentre erano entrambi offline.
Nel prossimo articolo della serie vediamo la stessa architettura in Node.js, questa volta con una persistenza vera (SQLite) su entrambi i lati.