package apiclient import ( "context" "crypto/tls" "errors" "log/slog" "net/http" "reflect" "github.com/AFASystems/presence/internal/pkg/common/appcontext" "github.com/AFASystems/presence/internal/pkg/config" "github.com/AFASystems/presence/internal/pkg/controller" "github.com/AFASystems/presence/internal/pkg/model" "github.com/segmentio/kafka-go" "gorm.io/gorm" "gorm.io/gorm/clause" ) func UpdateDB(db *gorm.DB, ctx context.Context, cfg *config.Config, writer *kafka.Writer, appState *appcontext.AppState) error { slog.Info("UpdateDB: Avvio procedura di sincronizzazione database...") tr := &http.Transport{ TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, } client := &http.Client{Transport: tr} token, err := GetToken(ctx, cfg, client) if err != nil { slog.Error("UpdateDB: Fallito recupero del Token di autenticazione", "error", err) return err } slog.Info("UpdateDB: Token di autenticazione recuperato con successo") // 1. Sincronizzazione TRACKERS trackers, err := GetTrackers(token, client, cfg) if err != nil { slog.Error("UpdateDB: Errore durante la chiamata GetTrackers", "error", err) } else { slog.Info("UpdateDB: Tracker scaricati dall'API", "count", len(trackers)) syncTable(db, trackers) slog.Info("UpdateDB: Tabella 'trackers' allineata su DB") if err := controller.SendKafkaMessage(writer, &model.ApiUpdate{Method: "DELETE", MAC: "all"}, ctx); err != nil { slog.Error("UpdateDB: Errore nell'invio del messaggio Kafka 'DELETE all' al lookup", "error", err) } kafkaMsgCount := 0 for _, v := range trackers { apiUpdate := model.ApiUpdate{ Method: "POST", ID: v.ID, MAC: v.MAC, } if err := controller.SendKafkaMessage(writer, &apiUpdate, ctx); err != nil { slog.Error("UpdateDB: Errore nell'invio del messaggio Kafka 'POST tracker'", "tracker_id", v.ID, "error", err) } else { kafkaMsgCount++ } } slog.Info("UpdateDB: Sincronizzazione messaggi Kafka per i tracker completata", "inviati", kafkaMsgCount) } // 2. Sincronizzazione GATEWAYS gateways, err := GetGateways(token, client, cfg) if err != nil { slog.Error("UpdateDB: Errore durante la chiamata GetGateways", "error", err) } else { slog.Info("UpdateDB: Gateway scaricati dall'API", "count", len(gateways)) syncTable(db, gateways) slog.Info("UpdateDB: Tabella 'gateways' allineata su DB") } // 3. Sincronizzazione FLOORS floors, err := GetFloors(token, client, cfg) if err != nil { slog.Error("UpdateDB: Errore durante la chiamata GetFloors", "error", err) } else { slog.Info("UpdateDB: Piani (floors) scaricati dall'API", "count", len(floors)) syncTable(db, floors) slog.Info("UpdateDB: Tabella 'floors' allineata su DB") } // 4. Sincronizzazione ZONES zones, err := GetZones(token, client, cfg) if err != nil { slog.Error("UpdateDB: Errore durante la chiamata GetZones", "error", err) } else { slog.Info("UpdateDB: Zone scaricate dall'API", "count", len(zones)) syncTable(db, zones) slog.Info("UpdateDB: Tabella 'zones' allineata su DB") } // 5. Sincronizzazione TRACKER ZONES trackerZones, err := GetTrackerZones(token, client, cfg) if err != nil { slog.Error("UpdateDB: Errore durante la chiamata GetTrackerZones", "error", err) } else { slog.Info("UpdateDB: Tracker-Zones scaricate dall'API", "count", len(trackerZones)) syncTable(db, trackerZones) slog.Info("UpdateDB: Tabella 'tracker_zones' allineata su DB") } // 6. Inferenza delle posizioni (Stime AI passate) inferredPosition, err := InferPosition(token, client, cfg) if err != nil { slog.Error("UpdateDB: Errore durante la chiamata InferPosition", "error", err) } else { updateCount := 0 for _, v := range inferredPosition.Items { mac := convertMac(v.Mac) res := db.Model(&model.Tracker{}).Where("mac = ?", mac).Updates(map[string]interface{}{"x": v.X, "y": v.Y}) if res.Error != nil { slog.Error("UpdateDB: Errore aggiornamento coordinate stimate per tracker", "mac", mac, "error", res.Error) } else if res.RowsAffected > 0 { updateCount++ } } slog.Info("UpdateDB: Aggiornamento posizioni storiche completato", "tracker_aggiornati", updateCount) } // 7. Controllo e Inizializzazione SETTINGS var settings appcontext.Settings if err := db.First(&settings).Error; err != nil { if !errors.Is(err, gorm.ErrRecordNotFound) { slog.Error("UpdateDB: Errore durante la lettura delle impostazioni", "error", err) } } if settings.ID == 0 { slog.Info("UpdateDB: Tabella settings vuota. Inizializzazione con i valori di default di AppState...") if err := db.Create(appState.GetSettings()).Error; err != nil { slog.Error("UpdateDB: Errore critico durante la creazione dei settings di default", "error", err) } else { slog.Info("UpdateDB: Settings di default salvati con successo") } } else { slog.Info("UpdateDB: Settings già presenti nel database", "id", settings.ID) } slog.Info("UpdateDB: Procedura di sincronizzazione completata.") return nil } func syncTable[T any](db *gorm.DB, data []T) { if len(data) == 0 { slog.Warn("syncTable: Nessun dato ricevuto dall'API, salto la sincronizzazione per questa tabella.") return } var ids []string for _, item := range data { v := reflect.ValueOf(item).FieldByName("ID").String() if v != "" { ids = append(ids, v) } } if len(ids) == 0 { slog.Error("syncTable: Impossibile procedere. Trovati 0 ID validi tramite reflection nella struttura dei dati.") return } err := db.Transaction(func(tx *gorm.DB) error { // Elimina i record locali che non esistono più nel backend centrale if err := tx.Where("id NOT IN ?", ids).Delete(new(T)).Error; err != nil { return err } // Upsert dei dati aggiornati o nuovi return tx.Clauses(clause.OnConflict{UpdateAll: true}).Create(&data).Error }) if err != nil { slog.Error("syncTable: Errore critico durante la transazione di sincronizzazione", "error", err) } }