package main import ( "bytes" "context" "encoding/json" "fmt" "io" "log" "math" "net/http" "strconv" "strings" "time" "unicode" "github.com/jackc/pgx/v5" "github.com/xuri/excelize/v2" ) // ============================================================================ // Innomaint API types and functions // ============================================================================ const ( innomaintAPIURL = "https://app.innomaint.com/api/public/api/restapicontroller/assetmanagement/getAssetIDDetailByAssetID" innomaintAPIToken = "QloRlKPbJCnHTe1DQNmU5JWcfv3FO9BgRtElU3LTeJI8FlQhnHFYFdbqFPJt" innomaintCustomerId = 17099 innomaintRateMS = 2000 // ms between requests — 0.5 req/sec avoids 429 innomaintMaxRetries = 3 innomaintRetryBase = 10 * time.Second // 10s, 20s, 40s backoff on 429 innomaintAPITimeout = 15 * time.Second innomaintConcurrency = 3 ) // InnomaintAPIResponse — top level type InnomaintAPIResponse struct { Status bool `json:"status"` ResponseCode int `json:"response_code"` Response InnomaintResponseBody `json:"response"` } type InnomaintResponseBody struct { Data InnomaintAssetData `json:"data"` Performance InnomaintPerformance `json:"performance"` WorkorderTransactional InnomaintWorkorder `json:"workorder_transactional"` AssetIDDetails InnomaintAssetIDDetails `json:"assetIDDeatails"` // note: typo in API } type InnomaintAssetData struct { ID int `json:"id"` TraceID int `json:"traceid"` Serialnumber string `json:"serialnumber"` Barcode string `json:"barcode"` Criticality string `json:"criticality"` CapacityRating string `json:"capacity_rating"` CustomerID int `json:"customer_id"` CustomerName string `json:"customer_name"` LocationID int `json:"location_id"` LocationName string `json:"location_name"` BuildingID int `json:"building_id"` BuildingName string `json:"building_name"` FloorID int `json:"floor_id"` FloorName string `json:"floor_name"` DepartmentID int `json:"department_id"` DepartmentName string `json:"department_name"` AssetFloorID int `json:"asset_floor_id"` AssetBuildingID int `json:"asset_building_id"` AssetDepartmentID int `json:"asset_department_id"` FkCustomersEquipmentMappingID int `json:"fk_customers_equipment_mapping_id"` CustomersLocationsSerialMappingID int `json:"customers_locations_serialnumber_mapping_id"` CustomerTraceabilityID int `json:"customer_traceability_id"` CustomerModelID int `json:"customer_model_id"` LocationTraceabilityID int `json:"location_traceability_id"` LocationModelID int `json:"location_model_id"` InChargePerson string `json:"in_charge_person"` Ticketcount int `json:"ticketcount"` Schedulecount int `json:"schedulecount"` ScheduleConfigured int `json:"schedule_configured"` ScheduleInitiate int `json:"schedule_initiate"` IotDeviceMapped int `json:"iot_device_mapped"` Status int `json:"status"` AssetLiveStatus int `json:"asset_live_status"` MappingStatus int `json:"mapping_status"` TraceabilityStatus int `json:"traceability_status"` TraceabilityUpdatedOn string `json:"traceability_updated_on"` InstallationDate *string `json:"installation_date"` YearOfManufacturing *int `json:"year_of_manufacturing"` IsLoaner int `json:"is_loaner"` IsBleTracking int `json:"is_ble_tracking"` IsRfidTracking int `json:"is_rfid_tracking"` IsQrTracking int `json:"is_qr_tracking"` IsTrackingEnabled int `json:"is_tracking_enabled"` IsAssetTracking int `json:"is_asset_tracking"` IsDeviceTracking int `json:"is_device_tracking"` AssetTrackingStatus int `json:"asset_tracking_status"` AssetReferenceNumber string `json:"asset_reference_number"` LoanerAvailability int `json:"loaner_availability"` SelectMachineOwner *int `json:"selectmachineowner"` SelectMachineOwnerName string `json:"selectmachineownername"` ContractName string `json:"contract_name"` ContractNumber string `json:"contract_number"` InContractType int `json:"in_contract_type"` ContractStatus int `json:"contract_status"` AssetContractID *int `json:"asset_contract_id"` AssetContractStartDate string `json:"asset_contract_start_date"` AssetContractEndDate string `json:"asset_contract_end_date"` IsExistingContract *int `json:"is_existing_contract"` AssetCurrentLattitude *string `json:"asset_current_lattitude"` AssetCurrentLongitude *string `json:"asset_current_longitude"` AssetCurrentLocation string `json:"asset_current_location"` BuildingGeoLocation string `json:"building_geo_location"` BuildingGeoLattitude string `json:"building_geo_lattitude"` BuildingGeoLongitude string `json:"building_geo_longitude"` GeoLocation string `json:"geo_location"` GeoLattitude *string `json:"geo_lattitude"` GeoLongitude *string `json:"geo_longitude"` LocationGeoAddress string `json:"location_geo_address"` LocationGeoLattitude *string `json:"location_geo_lattitude"` LocationGeoLongitude *string `json:"location_geo_longitude"` CustomerLattitude *string `json:"customer_lattitude"` CustomerLongitude *string `json:"customer_longitude"` CustomerGeoAddress *string `json:"customer_geo_address"` } type InnomaintPerformance struct { EquipmentsName string `json:"equipments_name"` EquipmentModelName string `json:"equipment_model_name"` CriticalityTypeLabel string `json:"criticality_type_label"` TotalTicket int `json:"total_ticket"` Mttr string `json:"mttr"` Mtbf string `json:"mtbf"` Availability string `json:"availability"` TotalTicketBreakdownHrs string `json:"total_ticket_breakdown_hrs"` TotalScheduleBreakdownHrs string `json:"total_schedule_breakdown_hrs"` AssetCreatedDate string `json:"asset_created_date"` TotalActualHrs string `json:"total_actual_hrs"` } type InnomaintWOPeriod struct { WorkEstimateCount string `json:"work_estimate_count"` WorkEstimateCost string `json:"work_estimate_cost"` ScheduleCount string `json:"schedule_count"` ScheduleCost string `json:"schedule_cost"` TicketCount string `json:"ticket_count"` TicketCost string `json:"ticket_cost"` } type InnomaintWorkorder struct { Upto InnomaintWOPeriod `json:"upto"` LastYear InnomaintWOPeriod `json:"last_year"` } type InnomaintAssetIDDetails struct { FkCompanyID int `json:"fk_company_id"` ModelID int `json:"model_id"` AssetID int `json:"asset_id"` ManufacturerID int `json:"manufacturer_id"` ManufacturerName string `json:"manufacturer_name"` CatID int `json:"cat_id"` CategoryName string `json:"category_name"` SubcategoryID int `json:"subcategory_id"` SubCategory string `json:"sub_category"` EquipmentsName string `json:"equipments_name"` EquipmentModelName string `json:"equipment_model_name"` EquipmentDescription string `json:"equipment_description"` IsMovable int `json:"is_movable"` AmcID int `json:"amc_id"` ContractType string `json:"contract_type"` IsMainAsset int `json:"is_main_asset"` IsFacility int `json:"is_facility"` ApprovalRequired int `json:"approval_required"` WorkorderRequired int `json:"workorder_required"` CustomerWorkorderApproval int `json:"customer_workorder_approval"` Qrcode string `json:"qrcode"` QrcodeImage string `json:"qrcode_image"` TraceabilityUpdatedOn string `json:"traceability_updated_on"` AssetBuildingID int `json:"asset_building_id"` AssetDepartmentID int `json:"asset_department_id"` BuildingGeoLattitude string `json:"building_geo_lattitude"` BuildingGeoLongitude string `json:"building_geo_longitude"` BuildingGeoLocation string `json:"building_geo_location"` CustomerID int `json:"customer_id"` CustomerName string `json:"customer_name"` LocationID int `json:"location_id"` LocationName string `json:"location_name"` BuildingID int `json:"building_id"` BuildingName string `json:"building_name"` FloorID int `json:"floor_id"` FloorName string `json:"floor_name"` DepartmentID int `json:"department_id"` DepartmentName string `json:"department_name"` } // SyncResult — returned by the sync job type SyncResult struct { Total int `json:"total"` Success int `json:"success"` Failed int `json:"failed"` Skipped int `json:"skipped"` Errors []string `json:"errors,omitempty"` Duration string `json:"duration_ms"` } // ============================================================================ // Data types // ============================================================================ // AssetRow holds one row from the Asset ID Report XLSX. // Column order (0-indexed, 19 columns): // // 0 S.No 1 Customer Name 2 Location 3 Manufacturer // 4 Category 5 Sub-Category 6 Asset Name 7 Asset Model // 8 Asset Criticality 9 Asset ID 10 Purchase Number 11 Purchase Date // 12 Purchase Value 13 Service Provider 14 Status 15 Reason Inactive // 16 Schedules Configured 17 Schedules Initiated 18 Schedule Assigned type AssetRow struct { SeqNo int CustomerName string Location string Manufacturer string Category string SubCategory string AssetName string AssetModel string AssetCriticality string AssetID string // report_asset_id → becomes ic3_asset_master.asset_id PurchaseNumber string PurchaseDate *time.Time PurchaseValue *float64 ServiceProviderName string Status int // 1=Active, 0=Inactive ReasonForInactive string ScheduleConfigured int ScheduleInitiated int ScheduleAssigned string } // ImportResult summarises one bulk import run. type ImportResult struct { Total int `json:"total"` Inserted int `json:"inserted"` Updated int `json:"updated"` Skipped int `json:"skipped"` Duration string `json:"duration_ms"` } // AssetRecord is the JSON shape for list/get endpoints. type AssetRecord struct { AssetID string `json:"asset_id"` AssetName string `json:"asset_name"` AssetType string `json:"asset_type"` AssetCategory *string `json:"asset_category"` AssetSubCategory *string `json:"asset_sub_category"` Criticality string `json:"criticality"` Traceid *int `json:"traceid"` CustomerName *string `json:"customer_name"` LocationName *string `json:"location_name"` BuildingName *string `json:"building_name"` SiteID *string `json:"site_id"` Department *string `json:"department"` Manufacturer *string `json:"manufacturer"` ModelNo *string `json:"model_no"` SerialNo *string `json:"serial_no"` CapacityRating *string `json:"capacity_rating"` AssetStatus string `json:"asset_status"` AssetLiveStatus int `json:"asset_live_status"` IotDeviceMapped int `json:"iot_device_mapped"` ScheduleConfigured int `json:"schedule_count_configured"` ScheduleInitiated int `json:"schedule_count_initiated"` Ticketcount int `json:"ticketcount"` PurchaseDate *time.Time `json:"purchase_date"` PurchaseValue *float64 `json:"purchase_value"` ContractEndDate *time.Time `json:"contract_end_date"` Latitude *float64 `json:"latitude"` Longitude *float64 `json:"longitude"` InstallDate *time.Time `json:"install_date"` LastSyncedAt *time.Time `json:"last_synced_at"` CreatedAt time.Time `json:"created_at"` } // ============================================================================ // XLSX parser // ============================================================================ // colIdx maps lower-cased header names to column indices for flexible xlsx parsing. type colIdx map[string]int func buildColIdx(header []string) colIdx { m := make(colIdx, len(header)) for i, h := range header { m[strings.ToLower(strings.TrimSpace(h))] = i } return m } // pick returns the first matching column index (-1 if none found). func (m colIdx) pick(names ...string) int { for _, n := range names { if i, ok := m[strings.ToLower(n)]; ok { return i } } return -1 } // ParseAssetReportXLSX reads the Asset ID Report from an io.Reader (xlsx format). // Column positions are detected from the header row, so any xlsx layout is supported. func ParseAssetReportXLSX(r io.Reader) ([]AssetRow, error) { f, err := excelize.OpenReader(r) if err != nil { return nil, fmt.Errorf("open xlsx: %w", err) } defer f.Close() sheets := f.GetSheetList() if len(sheets) == 0 { return nil, fmt.Errorf("xlsx has no sheets") } rows, err := f.GetRows(sheets[0]) if err != nil { return nil, fmt.Errorf("read rows: %w", err) } if len(rows) == 0 { return nil, fmt.Errorf("xlsx has no rows") } cm := buildColIdx(rows[0]) var out []AssetRow for i, cols := range rows { if i == 0 { continue } r := parseAssetRow(cols, cm) if r.AssetID == "" { continue } out = append(out, r) } return out, nil } func parseAssetRow(cols []string, cm colIdx) AssetRow { get := func(i int) string { if i < 0 || i >= len(cols) { return "" } v := strings.TrimSpace(cols[i]) if v == "-" || strings.EqualFold(v, "na") || strings.EqualFold(v, "n/a") { return "" } return v } col := func(names ...string) string { return get(cm.pick(names...)) } colInt := func(names ...string) int { n, _ := strconv.Atoi(col(names...)) return n } seqNo, _ := strconv.Atoi(col("s.no", "s. no", "s no", "sno")) r := AssetRow{ SeqNo: seqNo, CustomerName: col("customer name"), Location: col("location"), Manufacturer: col("manufacturer"), Category: normaliseCategory(col("category")), SubCategory: col("sub-category", "sub category", "subcategory"), AssetName: col("asset name"), AssetModel: col("asset model"), AssetCriticality: col("asset criticality", "criticality"), AssetID: col("asset id"), PurchaseNumber: col("purchase number"), ServiceProviderName: col("service provider name", "service provider"), ReasonForInactive: col("reason for inactive", "reason inactive"), ScheduleConfigured: colInt("no. of schedule configured", "schedules configured"), ScheduleInitiated: colInt("no. of schedule initiate", "schedules initiated"), ScheduleAssigned: col("schedule assigned"), } if strings.EqualFold(col("status"), "active") { r.Status = 1 } if ds := col("purchase date"); ds != "" { if t := parseFlexDate(ds); t != nil { r.PurchaseDate = t } } if vs := col("purchase value"); vs != "" { clean := strings.Map(func(c rune) rune { if unicode.IsDigit(c) || c == '.' { return c } return -1 }, vs) if v, err := strconv.ParseFloat(clean, 64); err == nil && v > 0 { r.PurchaseValue = &v } } return r } // normaliseCategory fixes dirty category values from the import. func normaliseCategory(s string) string { switch strings.ToUpper(strings.TrimSpace(s)) { case "INSTRUMENTATION", "INSTRUMENATION", "INSTRUMENENTATION": return "Instrument" case "PRESSURE GAUGE": return "Instrument" case "SAMPLE CATEGORY", "": return "Uncategorised" default: return s } } var dateFormats = []string{ "02-01-2006", "02/01/2006", "2006-01-02", "01/02/2006", "Jan 2, 2006", "2 Jan 2006", } func parseFlexDate(s string) *time.Time { for _, layout := range dateFormats { if t, err := time.Parse(layout, s); err == nil { return &t } } return nil } // ============================================================================ // DB — bulk import into ic3_asset_master (THE single asset table) // ============================================================================ // BulkImportAssets upserts rows into ic3_asset_master. // Conflict key: asset_id (= report Asset ID column, unique per asset). // replace=true truncates the table before inserting. // BulkImportAssets upserts all master tables then ic3_asset_master. // Insertion order respects FK dependencies: // // ic3_customer_master → ic3_location_master → ic3_manufacturer // → ic3_asset_category → ic3_asset_group → ic3_equipment_master // → ic3_asset_master // // replace=true truncates ic3_asset_master CASCADE before inserting. func (db *DB) BulkImportAssets(ctx context.Context, rows []AssetRow, replace bool) (ImportResult, error) { start := time.Now() res := ImportResult{Total: len(rows)} tx, err := db.pool.Begin(ctx) if err != nil { return res, fmt.Errorf("begin tx: %w", err) } defer tx.Rollback(ctx) if replace { if _, err := tx.Exec(ctx, `TRUNCATE ic3_asset_master CASCADE`); err != nil { return res, fmt.Errorf("truncate: %w", err) } } // Step 1: upsert all lookup/master tables, collect ID maps customerIDs, err := upsertCustomers(ctx, tx, rows) if err != nil { return res, fmt.Errorf("upsert customers: %w", err) } locationIDs, err := upsertLocations(ctx, tx, rows, customerIDs) if err != nil { return res, fmt.Errorf("upsert locations: %w", err) } manufacturerIDs, err := upsertManufacturers(ctx, tx, rows) if err != nil { return res, fmt.Errorf("upsert manufacturers: %w", err) } categoryIDs, err := upsertCategories(ctx, tx, rows) if err != nil { return res, fmt.Errorf("upsert categories: %w", err) } groupIDs, err := upsertAssetGroups(ctx, tx, rows, categoryIDs) if err != nil { return res, fmt.Errorf("upsert asset groups: %w", err) } equipmentIDs, err := upsertEquipment(ctx, tx, rows, categoryIDs, groupIDs, manufacturerIDs) if err != nil { return res, fmt.Errorf("upsert equipment: %w", err) } // Step 2: upsert ic3_asset_master in chunks of 500 const chunkSize = 500 for start2 := 0; start2 < len(rows); start2 += chunkSize { end := start2 + chunkSize if end > len(rows) { end = len(rows) } chunk := rows[start2:end] batch := &pgx.Batch{} for _, r := range chunk { custID := customerIDs[r.CustomerName] locID := locationIDs[r.Location] catID := categoryIDs[r.Category] grpID := groupIDs[r.SubCategory] eqID := equipmentIDs[equipKey(r.AssetName, r.AssetModel)] batch.Queue(` INSERT INTO ic3_asset_master ( serialnumber, cmms_asset_id, source_system_id, customer_id, location_id, asset_cat_id, asset_group_id, model_id, criticality, status, asset_status, asset_live_status, schedule_configured, schedule_initiate, schedule_assigned, schedulecount, ticketcount, po_number, purchase_date, purchase_value, service_provider_name, inactive_reason, report_seq_no, data_source, last_synced_at, created_at, updated_at ) VALUES ( $1,$2,$3,$4,$5,$6,$7,$8,$9,$10, $11,$12,$13,$14,$15,$16,$17,$18,$19,$20, $21,$22,$23,$24,NOW(),NOW(),NOW() ) ON CONFLICT (cmms_asset_id) DO UPDATE SET serialnumber = EXCLUDED.serialnumber, customer_id = EXCLUDED.customer_id, location_id = EXCLUDED.location_id, asset_cat_id = EXCLUDED.asset_cat_id, asset_group_id = EXCLUDED.asset_group_id, model_id = EXCLUDED.model_id, criticality = EXCLUDED.criticality, status = EXCLUDED.status, asset_live_status = EXCLUDED.asset_live_status, schedule_configured = EXCLUDED.schedule_configured, schedule_initiate = EXCLUDED.schedule_initiate, schedule_assigned = EXCLUDED.schedule_assigned, po_number = EXCLUDED.po_number, purchase_date = EXCLUDED.purchase_date, purchase_value = EXCLUDED.purchase_value, service_provider_name = EXCLUDED.service_provider_name, inactive_reason = EXCLUDED.inactive_reason, sync_version = COALESCE(ic3_asset_master.sync_version, 0) + 1, last_synced_at = NOW(), updated_at = NOW()`, r.AssetID, // $1 serialnumber (Innomaint Asset ID) r.AssetID, // $2 cmms_asset_id (Innomaint Asset ID) "INNOMAINT", // $3 source_system_id nullInt(custID), // $4 customer_id nullInt(locID), // $5 location_id nullInt(catID), // $6 asset_cat_id nullInt(grpID), // $7 asset_group_id nullInt(eqID), // $8 model_id r.AssetCriticality, // $9 criticality r.Status, // $10 status r.Status, // $11 asset_status 1, // $12 asset_live_status r.ScheduleConfigured, // $13 schedule_configured r.ScheduleInitiated, // $14 schedule_initiate nullStr(r.ScheduleAssigned), // $15 schedule_assigned r.ScheduleConfigured, // $16 schedulecount 0, // $17 ticketcount nullStr(r.PurchaseNumber), // $18 po_number r.PurchaseDate, // $19 purchase_date r.PurchaseValue, // $20 purchase_value nullStr(r.ServiceProviderName), // $21 service_provider_name nullStr(r.ReasonForInactive), // $22 inactive_reason r.SeqNo, // $23 report_seq_no "IMPORT", // $24 data_source ) } br := tx.SendBatch(ctx, batch) for range chunk { tag, err := br.Exec() if err != nil { res.Skipped++ log.Printf("BulkImportAssets row skip: %v", err) continue } if tag.RowsAffected() == 1 { res.Inserted++ } else { res.Updated++ } } if err := br.Close(); err != nil { return res, fmt.Errorf("batch close: %w", err) } } // Step 3: refresh aggregation counts if err := refreshAggregations(ctx, tx); err != nil { return res, fmt.Errorf("refresh aggregations: %w", err) } // Step 4: log to cmms_sync_log tx.Exec(ctx, ` INSERT INTO cmms_sync_log (sync_type, status, assets_synced, duration_ms) VALUES ('bulk_import','success',$1,$2)`, res.Inserted+res.Updated, int(time.Since(start).Milliseconds()), ) if err := tx.Commit(ctx); err != nil { return res, fmt.Errorf("commit: %w", err) } res.Duration = strconv.FormatInt(time.Since(start).Milliseconds(), 10) return res, nil } // ============================================================================ // BulkImportAssets helpers — upsert lookup tables and collect ID maps // ============================================================================ func upsertCustomers(ctx context.Context, tx pgx.Tx, rows []AssetRow) (map[string]int, error) { seen := map[string]struct{}{} for _, r := range rows { if r.CustomerName != "" { seen[r.CustomerName] = struct{}{} } } ids := map[string]int{} for name := range seen { var id int err := tx.QueryRow(ctx, ` INSERT INTO ic3_customer_master (customer_name, customer_status, source_system_id, created_at, updated_at) VALUES ($1, 1, 'INNOMAINT', NOW(), NOW()) ON CONFLICT (customer_name) DO UPDATE SET updated_at = NOW() RETURNING customer_id`, name).Scan(&id) if err != nil { return nil, fmt.Errorf("customer %q: %w", name, err) } ids[name] = id } return ids, nil } func upsertLocations(ctx context.Context, tx pgx.Tx, rows []AssetRow, customerIDs map[string]int) (map[string]int, error) { type locKey struct{ loc, cust string } seen := map[locKey]struct{}{} for _, r := range rows { if r.Location != "" { seen[locKey{r.Location, r.CustomerName}] = struct{}{} } } ids := map[string]int{} for k := range seen { custID := nullInt(customerIDs[k.cust]) var id int err := tx.QueryRow(ctx, ` INSERT INTO ic3_location_master (location_name, customer_id, location_status, source_system_id, created_at, updated_at) VALUES ($1, $2, 1, 'INNOMAINT', NOW(), NOW()) ON CONFLICT (location_name) DO UPDATE SET customer_id = EXCLUDED.customer_id, updated_at = NOW() RETURNING location_id`, k.loc, custID).Scan(&id) if err != nil { return nil, fmt.Errorf("location %q: %w", k.loc, err) } ids[k.loc] = id } return ids, nil } func upsertManufacturers(ctx context.Context, tx pgx.Tx, rows []AssetRow) (map[string]int, error) { seen := map[string]struct{}{} for _, r := range rows { if r.Manufacturer != "" { seen[r.Manufacturer] = struct{}{} } } ids := map[string]int{} for name := range seen { var id int err := tx.QueryRow(ctx, ` INSERT INTO ic3_manufacturer (manufacturer_name, manufacturer_status, created_at) VALUES ($1, 1, NOW()) ON CONFLICT (manufacturer_name) DO UPDATE SET manufacturer_status = 1 RETURNING manufacturer_id`, name).Scan(&id) if err != nil { return nil, fmt.Errorf("manufacturer %q: %w", name, err) } ids[name] = id } return ids, nil } func upsertCategories(ctx context.Context, tx pgx.Tx, rows []AssetRow) (map[string]int, error) { seen := map[string]struct{}{} for _, r := range rows { cat := r.Category if cat == "" { cat = "Uncategorised" } seen[cat] = struct{}{} } ids := map[string]int{} for name := range seen { var id int err := tx.QueryRow(ctx, ` INSERT INTO ic3_asset_category (category_name, category_status, created_at) VALUES ($1, 1, NOW()) ON CONFLICT (category_name) DO UPDATE SET category_status = 1 RETURNING cat_id`, name).Scan(&id) if err != nil { return nil, fmt.Errorf("category %q: %w", name, err) } ids[name] = id } return ids, nil } func upsertAssetGroups(ctx context.Context, tx pgx.Tx, rows []AssetRow, categoryIDs map[string]int) (map[string]int, error) { type grpKey struct{ sub, cat string } seen := map[grpKey]struct{}{} for _, r := range rows { if r.SubCategory != "" { seen[grpKey{r.SubCategory, r.Category}] = struct{}{} } } ids := map[string]int{} for k := range seen { catID := nullInt(categoryIDs[k.cat]) var id int err := tx.QueryRow(ctx, ` INSERT INTO ic3_asset_group (asset_group_name, cat_id, asset_group_status, created_at) VALUES ($1, $2, 1, NOW()) ON CONFLICT (asset_group_name) DO UPDATE SET cat_id = EXCLUDED.cat_id RETURNING subcategory_id`, k.sub, catID).Scan(&id) if err != nil { return nil, fmt.Errorf("asset group %q: %w", k.sub, err) } ids[k.sub] = id } return ids, nil } func equipKey(name, model string) string { return name + "||" + model } func upsertEquipment(ctx context.Context, tx pgx.Tx, rows []AssetRow, categoryIDs, groupIDs, manufacturerIDs map[string]int, ) (map[string]int, error) { type eqEntry struct{ name, model, cat, sub, mfr string } seen := map[string]eqEntry{} for _, r := range rows { k := equipKey(r.AssetName, r.AssetModel) if _, ok := seen[k]; !ok { seen[k] = eqEntry{r.AssetName, r.AssetModel, r.Category, r.SubCategory, r.Manufacturer} } } ids := map[string]int{} for k, e := range seen { catID := nullInt(categoryIDs[e.cat]) grpID := nullInt(groupIDs[e.sub]) mfrID := nullInt(manufacturerIDs[e.mfr]) modelName := e.model if modelName == "" { modelName = "NA" } var id int err := tx.QueryRow(ctx, ` INSERT INTO ic3_equipment_master ( equipments_name, equipment_model_name, equipment_description, cat_id, asset_group_id, manufacturer_id, manufacturer_name, model_status, created_at ) VALUES ($1,$2,$3,$4,$5,$6,$7,1,NOW()) ON CONFLICT (equipments_name, equipment_model_name) DO UPDATE SET cat_id = EXCLUDED.cat_id, asset_group_id = EXCLUDED.asset_group_id, manufacturer_id = EXCLUDED.manufacturer_id RETURNING model_id`, e.name, modelName, e.name, catID, grpID, mfrID, nullStr(e.mfr), ).Scan(&id) if err != nil { return nil, fmt.Errorf("equipment %q: %w", k, err) } ids[k] = id } return ids, nil } func refreshAggregations(ctx context.Context, tx pgx.Tx) error { if _, err := tx.Exec(ctx, `TRUNCATE ic3_asset_group_aggregation`); err != nil { return err } _, err := tx.Exec(ctx, ` INSERT INTO ic3_asset_group_aggregation ( group_id, group_name, cat_id, category_name, equipment_count, active_equipment, inactive_equipment, updated_at ) SELECT ag.subcategory_id, ag.asset_group_name, ag.cat_id, MAX(ac.category_name), COUNT(am.asset_id), SUM(CASE WHEN am.status = 1 THEN 1 ELSE 0 END), SUM(CASE WHEN am.status = 0 THEN 1 ELSE 0 END), NOW() FROM ic3_asset_group ag LEFT JOIN ic3_asset_category ac ON ac.cat_id = ag.cat_id LEFT JOIN ic3_asset_master am ON am.asset_group_id = ag.subcategory_id GROUP BY ag.subcategory_id, ag.asset_group_name, ag.cat_id`) return err } // ============================================================================ // DB — queries on ic3_asset_master // ============================================================================ func (db *DB) ListAssets(ctx context.Context, customer, category, status string, page, pageSize int) ([]AssetRecord, int, error) { if pageSize <= 0 || pageSize > 500 { pageSize = 100 } if page < 1 { page = 1 } offset := (page - 1) * pageSize where := []string{"1=1"} args := []any{} n := 1 if customer != "" { where = append(where, fmt.Sprintf("customer_name ILIKE $%d", n)) args = append(args, "%"+customer+"%") n++ } if category != "" { where = append(where, fmt.Sprintf("(asset_category ILIKE $%d OR asset_sub_category ILIKE $%d)", n, n)) args = append(args, "%"+category+"%") n++ } switch status { case "active": where = append(where, "asset_status = 'Active'") case "inactive": where = append(where, "asset_status = 'Inactive'") } cond := strings.Join(where, " AND ") var total int db.pool.QueryRow(ctx, `SELECT COUNT(*) FROM ic3_asset_master WHERE `+cond, args...).Scan(&total) args = append(args, pageSize, offset) q := fmt.Sprintf(` SELECT asset_id, asset_name, asset_type, asset_category, asset_sub_category, criticality, traceid, customer_name, location_name, building_name, site_id, department, manufacturer, model_no, serial_no, capacity_rating, asset_status, asset_live_status, iot_device_mapped, schedule_count_configured, schedule_count_initiated, ticketcount, purchase_date, purchase_value, contract_end_date, latitude, longitude, install_date, last_synced_at, created_at FROM ic3_asset_master WHERE %s ORDER BY report_seq_no ASC NULLS LAST, asset_id ASC LIMIT $%d OFFSET $%d`, cond, n, n+1) rows, err := db.pool.Query(ctx, q, args...) if err != nil { return nil, 0, err } defer rows.Close() var out []AssetRecord for rows.Next() { var a AssetRecord if err := rows.Scan( &a.AssetID, &a.AssetName, &a.AssetType, &a.AssetCategory, &a.AssetSubCategory, &a.Criticality, &a.Traceid, &a.CustomerName, &a.LocationName, &a.BuildingName, &a.SiteID, &a.Department, &a.Manufacturer, &a.ModelNo, &a.SerialNo, &a.CapacityRating, &a.AssetStatus, &a.AssetLiveStatus, &a.IotDeviceMapped, &a.ScheduleConfigured, &a.ScheduleInitiated, &a.Ticketcount, &a.PurchaseDate, &a.PurchaseValue, &a.ContractEndDate, &a.Latitude, &a.Longitude, &a.InstallDate, &a.LastSyncedAt, &a.CreatedAt, ); err != nil { continue } out = append(out, a) } return out, total, nil } func (db *DB) GetAsset(ctx context.Context, assetID string) (*AssetRecord, error) { var a AssetRecord err := db.pool.QueryRow(ctx, ` SELECT asset_id, asset_name, asset_type, asset_category, asset_sub_category, criticality, traceid, customer_name, location_name, building_name, site_id, department, manufacturer, model_no, serial_no, capacity_rating, asset_status, asset_live_status, iot_device_mapped, schedule_count_configured, schedule_count_initiated, ticketcount, purchase_date, purchase_value, contract_end_date, latitude, longitude, install_date, last_synced_at, created_at FROM ic3_asset_master WHERE asset_id = $1`, assetID, ).Scan( &a.AssetID, &a.AssetName, &a.AssetType, &a.AssetCategory, &a.AssetSubCategory, &a.Criticality, &a.Traceid, &a.CustomerName, &a.LocationName, &a.BuildingName, &a.SiteID, &a.Department, &a.Manufacturer, &a.ModelNo, &a.SerialNo, &a.CapacityRating, &a.AssetStatus, &a.AssetLiveStatus, &a.IotDeviceMapped, &a.ScheduleConfigured, &a.ScheduleInitiated, &a.Ticketcount, &a.PurchaseDate, &a.PurchaseValue, &a.ContractEndDate, &a.Latitude, &a.Longitude, &a.InstallDate, &a.LastSyncedAt, &a.CreatedAt, ) if err != nil { return nil, err } return &a, nil } // AssetHealthSummary is returned by GET /api/assets/health. type AssetHealthSummary struct { TotalAssets int `json:"total_assets"` ActiveAssets int `json:"active_assets"` InactiveAssets int `json:"inactive_assets"` NoIotLink int `json:"no_iot_link"` NoGPS int `json:"no_gps"` PMOverdue int `json:"pm_overdue"` PMDueSoon int `json:"pm_due_soon_7d"` ContractExpired int `json:"contract_expired"` ByCategory []map[string]any `json:"by_category"` ByCustomer []map[string]any `json:"by_customer"` DataQuality map[string]any `json:"data_quality"` LastImport *time.Time `json:"last_import"` } func (db *DB) GetAssetHealth(ctx context.Context) (AssetHealthSummary, error) { var h AssetHealthSummary db.pool.QueryRow(ctx, ` SELECT COUNT(*), SUM(CASE WHEN asset_status='Active' THEN 1 ELSE 0 END), SUM(CASE WHEN asset_status='Inactive' THEN 1 ELSE 0 END) FROM ic3_asset_master`).Scan(&h.TotalAssets, &h.ActiveAssets, &h.InactiveAssets) db.pool.QueryRow(ctx, `SELECT COUNT(*) FROM ic3_asset_master WHERE asset_status='Active' AND iot_device_mapped=0`).Scan(&h.NoIotLink) db.pool.QueryRow(ctx, `SELECT COUNT(*) FROM ic3_asset_master WHERE asset_status='Active' AND latitude IS NULL`).Scan(&h.NoGPS) db.pool.QueryRow(ctx, ` SELECT COUNT(*) FROM cmms_pm_schedule s JOIN ic3_asset_master m ON m.asset_id = s.asset_id WHERE m.asset_status='Active' AND s.is_overdue=TRUE`).Scan(&h.PMOverdue) db.pool.QueryRow(ctx, ` SELECT COUNT(*) FROM cmms_pm_schedule s JOIN ic3_asset_master m ON m.asset_id = s.asset_id WHERE m.asset_status='Active' AND s.next_schedule_date BETWEEN NOW() AND NOW()+INTERVAL '7 days'`).Scan(&h.PMDueSoon) db.pool.QueryRow(ctx, ` SELECT COUNT(*) FROM ic3_asset_master WHERE asset_status='Active' AND contract_end_date IS NOT NULL AND contract_end_date < NOW()`).Scan(&h.ContractExpired) catRows, _ := db.pool.Query(ctx, ` SELECT COALESCE(asset_category,'Unknown'), COALESCE(asset_sub_category,''), COUNT(*), SUM(CASE WHEN asset_status='Active' THEN 1 ELSE 0 END) FROM ic3_asset_master GROUP BY asset_category, asset_sub_category ORDER BY COUNT(*) DESC LIMIT 20`) if catRows != nil { for catRows.Next() { var cat, sub string var tot, active int catRows.Scan(&cat, &sub, &tot, &active) h.ByCategory = append(h.ByCategory, map[string]any{ "category": cat, "sub_category": sub, "total": tot, "active": active, }) } catRows.Close() } custRows, _ := db.pool.Query(ctx, ` SELECT COALESCE(customer_name,'Unknown'), COUNT(*), SUM(CASE WHEN asset_status='Active' THEN 1 ELSE 0 END) FROM ic3_asset_master GROUP BY customer_name ORDER BY COUNT(*) DESC LIMIT 20`) if custRows != nil { for custRows.Next() { var cust string var tot, active int custRows.Scan(&cust, &tot, &active) h.ByCustomer = append(h.ByCustomer, map[string]any{ "customer": cust, "total": tot, "active": active, }) } custRows.Close() } var avgQ, p1, p2 float64 db.pool.QueryRow(ctx, ` SELECT COALESCE(AVG(data_completeness_pct),0), COALESCE(SUM(gap_no_iot_link+gap_no_install_date+gap_no_gps+gap_no_contract),0), COALESCE(SUM(gap_no_purchase_date+gap_no_purchase_value+gap_no_lifespan+gap_no_depreciation+gap_no_vendor),0) FROM v_cmms_data_quality`).Scan(&avgQ, &p1, &p2) h.DataQuality = map[string]any{ "avg_completeness_pct": math.Round(avgQ*10) / 10, "total_p1_gaps": int(p1), "total_p2_gaps": int(p2), } var lastImport time.Time if err := db.pool.QueryRow(ctx, ` SELECT synced_at FROM cmms_sync_log WHERE sync_type='bulk_import' AND status='success' ORDER BY synced_at DESC LIMIT 1`).Scan(&lastImport); err == nil { h.LastImport = &lastImport } return h, nil } func (db *DB) GetPMOverdue(ctx context.Context) ([]map[string]any, error) { rows, err := db.pool.Query(ctx, ` SELECT asset_id, traceid, customer_name, location_name, building_name, criticality, in_charge_person, maintenance_name, next_schedule_date, schedule_reference_id, overdue_days, pm_urgency FROM v_cmms_pm_overdue LIMIT 500`) if err != nil { return nil, err } defer rows.Close() var out []map[string]any for rows.Next() { var ( aid, cust, loc, bldg, crit, inCharge, pmName, ref, urgency *string traceid, overdueDays *int nextDate *time.Time ) if err := rows.Scan(&aid, &traceid, &cust, &loc, &bldg, &crit, &inCharge, &pmName, &nextDate, &ref, &overdueDays, &urgency); err != nil { continue } out = append(out, map[string]any{ "asset_id": aid, "traceid": traceid, "customer_name": cust, "location_name": loc, "building_name": bldg, "criticality": crit, "in_charge_person": inCharge, "maintenance_name": pmName, "next_schedule_date": nextDate, "schedule_reference": ref, "overdue_days": overdueDays, "pm_urgency": urgency, }) } if out == nil { out = []map[string]any{} } return out, nil } // ============================================================================ // DB — Create / Update / Delete on ic3_asset_master // ============================================================================ // AssetUpsertInput is the shape accepted by Create and Update. type AssetUpsertInput struct { AssetID string `json:"asset_id"` AssetName string `json:"asset_name"` AssetType string `json:"asset_type"` AssetCategory string `json:"asset_category"` AssetSubCategory string `json:"asset_sub_category"` Criticality string `json:"criticality"` CustomerName string `json:"customer_name"` LocationName string `json:"location_name"` BuildingName string `json:"building_name"` SiteID string `json:"site_id"` Department string `json:"department"` InChargePerson string `json:"in_charge_person"` Manufacturer string `json:"manufacturer"` ModelNo string `json:"model_no"` SerialNo string `json:"serial_no"` CapacityRating string `json:"capacity_rating"` AssetStatus string `json:"asset_status"` Latitude *float64 `json:"latitude"` Longitude *float64 `json:"longitude"` InstallDate string `json:"install_date"` PurchaseValue *float64 `json:"purchase_value"` ContractEndDate string `json:"contract_end_date"` DesignLifeYears *int `json:"design_life_years"` } func (db *DB) CreateAsset(ctx context.Context, in AssetUpsertInput) (*AssetRecord, error) { if in.AssetID == "" || in.AssetName == "" { return nil, fmt.Errorf("asset_id and asset_name are required") } if in.AssetStatus == "" { in.AssetStatus = "Active" } if in.Criticality == "" { in.Criticality = "Non-Critical" } _, err := db.pool.Exec(ctx, ` INSERT INTO ic3_asset_master ( asset_id, asset_name, asset_type, asset_category, asset_sub_category, criticality, customer_name, location_name, building_name, site_id, department, in_charge_person, manufacturer, model_no, serial_no, capacity_rating, asset_status, latitude, longitude, install_date, purchase_value, contract_end_date, design_life_years, data_source, created_at, updated_at ) VALUES ( $1,$2,$3,$4,$5,$6,$7,$8,$9,$10, $11,$12,$13,$14,$15,$16,$17,$18,$19, $20,$21,$22,$23,'MANUAL',NOW(),NOW() )`, in.AssetID, in.AssetName, nullStr(in.AssetType), nullStr(in.AssetCategory), nullStr(in.AssetSubCategory), in.Criticality, nullStr(in.CustomerName), nullStr(in.LocationName), nullStr(in.BuildingName), nullStr(in.SiteID), nullStr(in.Department), nullStr(in.InChargePerson), nullStr(in.Manufacturer), nullStr(in.ModelNo), nullStr(in.SerialNo), nullStr(in.CapacityRating), in.AssetStatus, in.Latitude, in.Longitude, nullStr(in.InstallDate), in.PurchaseValue, nullStr(in.ContractEndDate), in.DesignLifeYears, ) if err != nil { return nil, err } return db.GetAsset(ctx, in.AssetID) } func (db *DB) UpdateAsset(ctx context.Context, assetID string, in AssetUpsertInput) (*AssetRecord, error) { if in.AssetName == "" { return nil, fmt.Errorf("asset_name is required") } if in.AssetStatus == "" { in.AssetStatus = "Active" } _, err := db.pool.Exec(ctx, ` UPDATE ic3_asset_master SET asset_name = $2, asset_type = $3, asset_category = $4, asset_sub_category= $5, criticality = $6, customer_name = $7, location_name = $8, building_name = $9, site_id = $10, department = $11, in_charge_person = $12, manufacturer = $13, model_no = $14, serial_no = $15, capacity_rating = $16, asset_status = $17, latitude = $18, longitude = $19, install_date = $20, purchase_value = $21, contract_end_date = $22, design_life_years = $23, updated_at = NOW() WHERE asset_id = $1`, assetID, in.AssetName, nullStr(in.AssetType), nullStr(in.AssetCategory), nullStr(in.AssetSubCategory), in.Criticality, nullStr(in.CustomerName), nullStr(in.LocationName), nullStr(in.BuildingName), nullStr(in.SiteID), nullStr(in.Department), nullStr(in.InChargePerson), nullStr(in.Manufacturer), nullStr(in.ModelNo), nullStr(in.SerialNo), nullStr(in.CapacityRating), in.AssetStatus, in.Latitude, in.Longitude, nullStr(in.InstallDate), in.PurchaseValue, nullStr(in.ContractEndDate), in.DesignLifeYears, ) if err != nil { return nil, err } return db.GetAsset(ctx, assetID) } func (db *DB) DeleteAsset(ctx context.Context, assetID string) error { tag, err := db.pool.Exec(ctx, `DELETE FROM ic3_asset_master WHERE asset_id = $1`, assetID) if err != nil { return err } if tag.RowsAffected() == 0 { return fmt.Errorf("not found") } return nil } func (db *DB) GetSyncLog(ctx context.Context, limit int) ([]map[string]any, error) { if limit <= 0 || limit > 200 { limit = 50 } rows, err := db.pool.Query(ctx, ` SELECT id, cmms_id, sync_type, status, assets_synced, response_code, error_msg, duration_ms, synced_at FROM cmms_sync_log ORDER BY synced_at DESC LIMIT $1`, limit) if err != nil { return nil, err } defer rows.Close() var out []map[string]any for rows.Next() { var id, assets int64 var cmmsID *int64 var syncType, st string var rc, dur *int var errMsg *string var syncedAt time.Time rows.Scan(&id, &cmmsID, &syncType, &st, &assets, &rc, &errMsg, &dur, &syncedAt) out = append(out, map[string]any{ "id": id, "cmms_id": cmmsID, "sync_type": syncType, "status": st, "assets_synced": assets, "response_code": rc, "error_msg": errMsg, "duration_ms": dur, "synced_at": syncedAt, }) } if out == nil { out = []map[string]any{} } return out, nil } // ============================================================================ // Innomaint API functions // ============================================================================ // FetchInnomaintAssetDetail calls the Innomaint API for one asset func FetchInnomaintAssetDetail(ctx context.Context, client *http.Client, serialnumber string) (*InnomaintAPIResponse, error) { body, _ := json.Marshal(map[string]string{"serialnumber": serialnumber}) req, err := http.NewRequestWithContext(ctx, "POST", innomaintAPIURL, bytes.NewReader(body)) if err != nil { return nil, err } req.Header.Set("Content-Type", "application/json") req.Header.Set("Authorization", "Bearer "+innomaintAPIToken) resp, err := client.Do(req) if err != nil { return nil, fmt.Errorf("http: %w", err) } defer resp.Body.Close() if resp.StatusCode != 200 { return nil, fmt.Errorf("API status %d", resp.StatusCode) } rawBody, _ := io.ReadAll(resp.Body) log.Printf("[Innomaint] %s HTTP=%d body_prefix=%.200s", serialnumber, resp.StatusCode, string(rawBody)) var result InnomaintAPIResponse if err := json.Unmarshal(rawBody, &result); err != nil { return nil, fmt.Errorf("decode: %w", err) } if !result.Status || result.ResponseCode != 200 { return nil, fmt.Errorf("API returned status=false (status=%v responseCode=%d)", result.Status, result.ResponseCode) } return &result, nil } // FetchInnomaintAssetDetailWithRetry wraps FetchInnomaintAssetDetail // with up to innomaintMaxRetries attempts using exponential backoff. func FetchInnomaintAssetDetailWithRetry(ctx context.Context, client *http.Client, serialnumber string) (*InnomaintAPIResponse, error) { var lastErr error for attempt := 0; attempt < innomaintMaxRetries; attempt++ { if attempt > 0 { // Exponential backoff: 1s, 2s, 4s wait := innomaintRetryBase * time.Duration(1<<(attempt-1)) log.Printf("Retry %d/%d for %q — waiting %s (last err: %v)", attempt, innomaintMaxRetries, serialnumber, wait, lastErr) select { case <-time.After(wait): case <-ctx.Done(): return nil, fmt.Errorf("context cancelled during retry wait: %w", ctx.Err()) } } // Per-call timeout callCtx, cancel := context.WithTimeout(ctx, innomaintAPITimeout) resp, err := FetchInnomaintAssetDetail(callCtx, client, serialnumber) cancel() if err == nil { if attempt > 0 { log.Printf("Retry succeeded for %q on attempt %d", serialnumber, attempt+1) } return resp, nil } lastErr = err // Don't retry on context cancellation if ctx.Err() != nil { return nil, fmt.Errorf("context cancelled: %w", ctx.Err()) } // Don't retry on 404 — asset doesn't exist in Innomaint if strings.Contains(err.Error(), "status 404") { return nil, fmt.Errorf("asset not found in innomaint (404): %w", err) } } return nil, fmt.Errorf("all %d attempts failed for %q: %w", innomaintMaxRetries, serialnumber, lastErr) } // EnrichAssetFromAPI updates ic3_asset_master + related tables from API response func (db *DB) EnrichAssetFromAPI(ctx context.Context, tx pgx.Tx, apiResp *InnomaintAPIResponse) error { d := apiResp.Response.Data p := apiResp.Response.Performance wo := apiResp.Response.WorkorderTransactional det := apiResp.Response.AssetIDDetails // ---------------------------------------------------------------- // 1. Upsert ic3_customer_master — store innomaint_customer_id + geo // ---------------------------------------------------------------- var localCustID *int if d.CustomerID > 0 && d.CustomerName != "" { var id int err := tx.QueryRow(ctx, ` INSERT INTO ic3_customer_master ( customer_name, customer_status, source_system_id, innomaint_customer_id, geo_lattitude, geo_longitude, geo_address, created_at, updated_at ) VALUES ($1, 1, 'INNOMAINT', $2, $3, $4, $5, NOW(), NOW()) ON CONFLICT (customer_name) DO UPDATE SET innomaint_customer_id = EXCLUDED.innomaint_customer_id, geo_lattitude = COALESCE(EXCLUDED.geo_lattitude, ic3_customer_master.geo_lattitude), geo_longitude = COALESCE(EXCLUDED.geo_longitude, ic3_customer_master.geo_longitude), geo_address = COALESCE(EXCLUDED.geo_address, ic3_customer_master.geo_address), updated_at = NOW() RETURNING customer_id`, d.CustomerName, d.CustomerID, d.CustomerLattitude, d.CustomerLongitude, d.CustomerGeoAddress, ).Scan(&id) if err != nil { return fmt.Errorf("upsert customer: %w", err) } localCustID = &id } else { tx.QueryRow(ctx, `SELECT customer_id FROM ic3_customer_master WHERE customer_name = $1 LIMIT 1`, d.CustomerName, ).Scan(&localCustID) } // ---------------------------------------------------------------- // 2. Upsert ic3_location_master — store innomaint_location_id + geo // ---------------------------------------------------------------- var localLocID *int if d.LocationID > 0 && d.LocationName != "" { var id int err := tx.QueryRow(ctx, ` INSERT INTO ic3_location_master ( location_name, customer_id, location_status, source_system_id, innomaint_location_id, geo_lattitude, geo_longitude, geo_address, created_at, updated_at ) VALUES ($1, $2, 1, 'INNOMAINT', $3, $4, $5, $6, NOW(), NOW()) ON CONFLICT (location_name) DO UPDATE SET innomaint_location_id = EXCLUDED.innomaint_location_id, customer_id = COALESCE(EXCLUDED.customer_id, ic3_location_master.customer_id), geo_lattitude = COALESCE(EXCLUDED.geo_lattitude, ic3_location_master.geo_lattitude), geo_longitude = COALESCE(EXCLUDED.geo_longitude, ic3_location_master.geo_longitude), geo_address = COALESCE(EXCLUDED.geo_address, ic3_location_master.geo_address), updated_at = NOW() RETURNING location_id`, d.LocationName, localCustID, d.LocationID, d.LocationGeoLattitude, d.LocationGeoLongitude, nullStr(d.LocationGeoAddress), ).Scan(&id) if err != nil { return fmt.Errorf("upsert location: %w", err) } localLocID = &id } else { tx.QueryRow(ctx, `SELECT location_id FROM ic3_location_master WHERE location_name = $1 LIMIT 1`, d.LocationName, ).Scan(&localLocID) } // ---------------------------------------------------------------- // 3. Upsert ic3_building — innomaint_building_id + geo lat/lng // ---------------------------------------------------------------- if d.BuildingID > 0 && d.BuildingName != "" { _, err := tx.Exec(ctx, ` INSERT INTO ic3_building ( building_id, customer_id, building_name, innomaint_building_id, geo_lattitude, geo_longitude, geo_location, created_at, updated_at ) VALUES ($1, $2, $3, $4, $5, $6, $7, NOW(), NOW()) ON CONFLICT (building_id) DO UPDATE SET building_name = EXCLUDED.building_name, innomaint_building_id = EXCLUDED.innomaint_building_id, geo_lattitude = COALESCE(EXCLUDED.geo_lattitude, ic3_building.geo_lattitude), geo_longitude = COALESCE(EXCLUDED.geo_longitude, ic3_building.geo_longitude), geo_location = COALESCE(EXCLUDED.geo_location, ic3_building.geo_location), updated_at = NOW()`, d.BuildingID, localCustID, d.BuildingName, d.BuildingID, // innomaint_building_id same as building_id from API nullStr(d.BuildingGeoLattitude), nullStr(d.BuildingGeoLongitude), nullStr(d.BuildingGeoLocation), ) if err != nil { return fmt.Errorf("upsert building: %w", err) } } // ---------------------------------------------------------------- // 4. Upsert ic3_floor_master — innomaint_floor_id // ---------------------------------------------------------------- if d.FloorID > 0 && d.FloorName != "" { _, err := tx.Exec(ctx, ` INSERT INTO ic3_floor_master ( floor_id, floor_name, location_id, asset_floor_id, innomaint_floor_id, created_at, updated_at ) VALUES ($1, $2, $3, $4, $5, NOW(), NOW()) ON CONFLICT (floor_id) DO UPDATE SET floor_name = EXCLUDED.floor_name, innomaint_floor_id = EXCLUDED.innomaint_floor_id, updated_at = NOW()`, d.FloorID, d.FloorName, localLocID, nullInt(d.AssetFloorID), d.FloorID, ) if err != nil { return fmt.Errorf("upsert floor: %w", err) } } // ---------------------------------------------------------------- // 5. Upsert ic3_department_master — innomaint_department_id // ---------------------------------------------------------------- if d.DepartmentID > 0 && d.DepartmentName != "" { _, err := tx.Exec(ctx, ` INSERT INTO ic3_department_master ( department_id, department_name, floor_id, location_id, asset_department_id, innomaint_department_id, created_at, updated_at ) VALUES ($1, $2, $3, $4, $5, $6, NOW(), NOW()) ON CONFLICT (department_id) DO UPDATE SET department_name = EXCLUDED.department_name, innomaint_department_id = EXCLUDED.innomaint_department_id, updated_at = NOW()`, d.DepartmentID, d.DepartmentName, nullInt(d.FloorID), localLocID, nullInt(d.AssetDepartmentID), d.DepartmentID, ) if err != nil { return fmt.Errorf("upsert department: %w", err) } } // ---------------------------------------------------------------- // 6. Upsert ic3_asset_category — innomaint_cat_id // ---------------------------------------------------------------- var localCatID *int if det.CatID > 0 && det.CategoryName != "" { var id int err := tx.QueryRow(ctx, ` INSERT INTO ic3_asset_category ( category_name, category_status, innomaint_cat_id, created_at ) VALUES ($1, 1, $2, NOW()) ON CONFLICT (category_name) DO UPDATE SET innomaint_cat_id = EXCLUDED.innomaint_cat_id RETURNING cat_id`, det.CategoryName, det.CatID, ).Scan(&id) if err != nil { return fmt.Errorf("upsert category: %w", err) } localCatID = &id } // ---------------------------------------------------------------- // 7. Upsert ic3_asset_group — innomaint_subcategory_id // ---------------------------------------------------------------- var localGroupID *int if det.SubcategoryID > 0 && det.SubCategory != "" { var id int err := tx.QueryRow(ctx, ` INSERT INTO ic3_asset_group ( asset_group_name, cat_id, asset_group_status, innomaint_subcategory_id, created_at ) VALUES ($1, $2, 1, $3, NOW()) ON CONFLICT (asset_group_name) DO UPDATE SET innomaint_subcategory_id = EXCLUDED.innomaint_subcategory_id, cat_id = COALESCE(EXCLUDED.cat_id, ic3_asset_group.cat_id) RETURNING subcategory_id`, det.SubCategory, localCatID, det.SubcategoryID, ).Scan(&id) if err != nil { return fmt.Errorf("upsert asset group: %w", err) } localGroupID = &id } // ---------------------------------------------------------------- // 8. Upsert ic3_manufacturer — innomaint_manufacturer_id // ---------------------------------------------------------------- var localMfrID *int if det.ManufacturerID > 0 && det.ManufacturerName != "" { var id int err := tx.QueryRow(ctx, ` INSERT INTO ic3_manufacturer ( manufacturer_name, manufacturer_status, innomaint_manufacturer_id, created_at ) VALUES ($1, 1, $2, NOW()) ON CONFLICT (manufacturer_name) DO UPDATE SET innomaint_manufacturer_id = EXCLUDED.innomaint_manufacturer_id RETURNING manufacturer_id`, det.ManufacturerName, det.ManufacturerID, ).Scan(&id) if err != nil { return fmt.Errorf("upsert manufacturer: %w", err) } localMfrID = &id } // ---------------------------------------------------------------- // 9. Upsert ic3_equipment_master — innomaint_model_id + description // ---------------------------------------------------------------- var localModelID *int if det.ModelID > 0 && det.EquipmentsName != "" { modelName := det.EquipmentModelName if modelName == "" { modelName = "NA" } var id int err := tx.QueryRow(ctx, ` INSERT INTO ic3_equipment_master ( equipments_name, equipment_model_name, equipment_description, cat_id, asset_group_id, manufacturer_id, manufacturer_name, innomaint_model_id, is_movable, model_status, created_at ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,1,NOW()) ON CONFLICT (equipments_name, equipment_model_name) DO UPDATE SET innomaint_model_id = EXCLUDED.innomaint_model_id, equipment_description = COALESCE(EXCLUDED.equipment_description, ic3_equipment_master.equipment_description), cat_id = COALESCE(EXCLUDED.cat_id, ic3_equipment_master.cat_id), asset_group_id = COALESCE(EXCLUDED.asset_group_id, ic3_equipment_master.asset_group_id), manufacturer_id = COALESCE(EXCLUDED.manufacturer_id, ic3_equipment_master.manufacturer_id), is_movable = EXCLUDED.is_movable RETURNING model_id`, det.EquipmentsName, modelName, nullStr(det.EquipmentDescription), localCatID, localGroupID, localMfrID, nullStr(det.ManufacturerName), det.ModelID, det.IsMovable, ).Scan(&id) if err != nil { return fmt.Errorf("upsert equipment: %w", err) } localModelID = &id } // ---------------------------------------------------------------- // 10. Parse traceability_updated_on timestamp // ---------------------------------------------------------------- var traceUpdatedOn *time.Time if det.TraceabilityUpdatedOn != "" { layouts := []string{ "2006-01-02 15:04:05+00", "2006-01-02 15:04:05", "2006-01-02T15:04:05Z", } for _, layout := range layouts { if t, err := time.Parse(layout, det.TraceabilityUpdatedOn); err == nil { traceUpdatedOn = &t break } } } // ---------------------------------------------------------------- // 11. UPDATE ic3_asset_master — full enrich with all fields + local FKs // ---------------------------------------------------------------- _, err := tx.Exec(ctx, ` UPDATE ic3_asset_master SET traceability_id = $2, barcode = $3, qrcode = $4, qrcode_image = $5, criticality = $6, capacity_rating = $7, asset_reference_number = $8, customer_id = COALESCE($9, customer_id), location_id = COALESCE($10, location_id), building_id = $11, floor_id = $12, department_id = $13, asset_building_id = $14, asset_floor_id = $15, asset_department_id = $16, model_id = COALESCE($17, model_id), asset_cat_id = COALESCE($18, asset_cat_id), asset_group_id = COALESCE($19, asset_group_id), fk_customers_equipment_mapping_id = $20, customers_locations_serialnumber_mapping_id = $21, customer_traceability_id = $22, customer_model_id = $23, location_traceability_id = $24, location_model_id = $25, in_charge_person = $26, ticketcount = $27, schedulecount = $28, schedule_configured = $29, schedule_initiate = $30, iot_device_mapped = $31, status = $32, asset_live_status = $33, mapping_status = $34, traceability_status = $35, is_loaner = $36, loaner_availability = $37, is_ble_tracking = $38, is_rfid_tracking = $39, is_qr_tracking = $40, is_tracking_enabled = $41, is_asset_tracking = $42, is_device_tracking = $43, asset_tracking_status = $44, is_main_asset = $45, is_facility = $46, fk_company_id = $47, approval_required = $48, workorder_required = $49, amc_id = $50, contract_type = $51, contract_name = $52, contract_number = $53, in_contract_type = $54, contract_status = $55, asset_contract_id = $56, asset_contract_start_date = $57, asset_contract_end_date = $58, is_existing_contract = $59, asset_current_lattitude = $60, asset_current_longitude = $61, asset_current_location = $62, traceability_updated_on = $63, last_synced_at = NOW(), sync_version = COALESCE(sync_version, 0) + 1, updated_at = NOW() WHERE cmms_asset_id = $1`, d.Serialnumber, // $1 WHERE key d.TraceID, // $2 nullStr(d.Barcode), // $3 nullStr(det.Qrcode), // $4 nullStr(det.QrcodeImage), // $5 nullStr(d.Criticality), // $6 nullStr(d.CapacityRating), // $7 nullStr(d.AssetReferenceNumber), // $8 localCustID, // $9 localLocID, // $10 nullInt(d.BuildingID), // $11 nullInt(d.FloorID), // $12 nullInt(d.DepartmentID), // $13 nullInt(det.AssetBuildingID), // $14 nullInt(d.AssetFloorID), // $15 nullInt(det.AssetDepartmentID), // $16 localModelID, // $17 localCatID, // $18 localGroupID, // $19 nullInt(d.FkCustomersEquipmentMappingID), // $20 nullInt(d.CustomersLocationsSerialMappingID), // $21 nullInt(d.CustomerTraceabilityID), // $22 nullInt(d.CustomerModelID), // $23 nullInt(d.LocationTraceabilityID), // $24 nullInt(d.LocationModelID), // $25 nullStr(d.InChargePerson), // $26 d.Ticketcount, // $27 d.Schedulecount, // $28 d.ScheduleConfigured, // $29 d.ScheduleInitiate, // $30 d.IotDeviceMapped, // $31 d.Status, // $32 d.AssetLiveStatus, // $33 d.MappingStatus, // $34 d.TraceabilityStatus, // $35 d.IsLoaner, // $36 d.LoanerAvailability, // $37 d.IsBleTracking, // $38 d.IsRfidTracking, // $39 d.IsQrTracking, // $40 d.IsTrackingEnabled, // $41 d.IsAssetTracking, // $42 d.IsDeviceTracking, // $43 d.AssetTrackingStatus, // $44 det.IsMainAsset, // $45 det.IsFacility, // $46 nullInt(det.FkCompanyID), // $47 det.ApprovalRequired, // $48 det.WorkorderRequired, // $49 nullStrFromInt(det.AmcID), // $50 — amc_id is TEXT column nullStr(det.ContractType), // $51 nullStr(d.ContractName), // $52 nullStr(d.ContractNumber), // $53 d.InContractType, // $54 d.ContractStatus, // $55 d.AssetContractID, // $56 nullStr(d.AssetContractStartDate), // $57 nullStr(d.AssetContractEndDate), // $58 d.IsExistingContract, // $59 d.AssetCurrentLattitude, // $60 d.AssetCurrentLongitude, // $61 nullStr(d.AssetCurrentLocation), // $62 traceUpdatedOn, // $63 ) if err != nil { return fmt.Errorf("update asset_master: %w", err) } // ---------------------------------------------------------------- // 12. Get local asset_id for child table inserts // ---------------------------------------------------------------- var assetID int err = tx.QueryRow(ctx, `SELECT asset_id FROM ic3_asset_master WHERE cmms_asset_id = $1`, d.Serialnumber, ).Scan(&assetID) if err != nil { log.Printf("EnrichAssetFromAPI: asset not found for cmms_asset_id=%q: %v", d.Serialnumber, err) return fmt.Errorf("get asset_id for %q: %w", d.Serialnumber, err) } // ---------------------------------------------------------------- // 13. Upsert ic3_asset_performance // ---------------------------------------------------------------- _, err = tx.Exec(ctx, ` INSERT INTO ic3_asset_performance ( asset_id, serialnumber, equipments_name, equipment_model_name, criticality_type_label, total_ticket, mttr, mtbf, availability, total_ticket_breakdown_hrs, total_schedule_breakdown_hrs, asset_created_date, total_actual_hrs, recorded_at ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,NOW()) ON CONFLICT (asset_id) DO UPDATE SET total_ticket = EXCLUDED.total_ticket, mttr = EXCLUDED.mttr, mtbf = EXCLUDED.mtbf, availability = EXCLUDED.availability, total_ticket_breakdown_hrs = EXCLUDED.total_ticket_breakdown_hrs, total_schedule_breakdown_hrs = EXCLUDED.total_schedule_breakdown_hrs, total_actual_hrs = EXCLUDED.total_actual_hrs, recorded_at = NOW()`, assetID, d.Serialnumber, p.EquipmentsName, p.EquipmentModelName, p.CriticalityTypeLabel, p.TotalTicket, p.Mttr, p.Mtbf, p.Availability, p.TotalTicketBreakdownHrs, p.TotalScheduleBreakdownHrs, p.AssetCreatedDate, p.TotalActualHrs, ) if err != nil { return fmt.Errorf("upsert performance: %w", err) } // ---------------------------------------------------------------- // 14. Upsert ic3_workorder_summary — upto + last_year // ---------------------------------------------------------------- for _, period := range []struct { pType string wop InnomaintWOPeriod }{ {"upto", wo.Upto}, {"last_year", wo.LastYear}, } { weCount, _ := strconv.Atoi(period.wop.WorkEstimateCount) weCost, _ := strconv.ParseFloat(period.wop.WorkEstimateCost, 64) scCount, _ := strconv.Atoi(period.wop.ScheduleCount) scCost, _ := strconv.ParseFloat(period.wop.ScheduleCost, 64) tkCount, _ := strconv.Atoi(period.wop.TicketCount) tkCost, _ := strconv.ParseFloat(period.wop.TicketCost, 64) _, err = tx.Exec(ctx, ` INSERT INTO ic3_workorder_summary ( asset_id, period_type, work_estimate_count, work_estimate_cost, schedule_count, schedule_cost, ticket_count, ticket_cost, updated_at ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,NOW()) ON CONFLICT (asset_id, period_type) DO UPDATE SET work_estimate_count = EXCLUDED.work_estimate_count, work_estimate_cost = EXCLUDED.work_estimate_cost, schedule_count = EXCLUDED.schedule_count, schedule_cost = EXCLUDED.schedule_cost, ticket_count = EXCLUDED.ticket_count, ticket_cost = EXCLUDED.ticket_cost, updated_at = NOW()`, assetID, period.pType, weCount, weCost, scCount, scCost, tkCount, tkCost, ) if err != nil { return fmt.Errorf("upsert workorder %s: %w", period.pType, err) } } return nil } // SyncAllAssetsFromInnomaint fetches all (or batched) assets from Innomaint API // and enriches ic3_asset_master + related tables. // batchSize: 0 = all assets // offsetStart: skip first N (for pagination/resume) // retryFailed: true = only sync assets where last ic3_sync_log status='error' func (db *DB) SyncAllAssetsFromInnomaint(ctx context.Context, batchSize, offsetStart int, retryFailed bool, customerID int) (SyncResult, error) { start := time.Now() result := SyncResult{} var assetIDs []string var err error if retryFailed { assetIDs, err = db.getFailedSyncAssets(ctx, batchSize, customerID) } else { assetIDs, err = db.getSyncAssetIDs(ctx, batchSize, offsetStart, customerID) } if err != nil { return result, fmt.Errorf("fetch asset ids: %w", err) } result.Total = len(assetIDs) if result.Total == 0 { result.Duration = "0" return result, nil } log.Printf("Starting sync: %d assets (offset=%d, retryFailed=%v, customerID=%d)", result.Total, offsetStart, retryFailed, customerID) client := &http.Client{ Timeout: innomaintAPITimeout, Transport: &http.Transport{ MaxIdleConns: 10, IdleConnTimeout: 30 * time.Second, }, } for i, cmmsID := range assetIDs { if ctx.Err() != nil { break } assetStart := time.Now() time.Sleep(time.Duration(innomaintRateMS) * time.Millisecond) apiResp, fetchErr := FetchInnomaintAssetDetailWithRetry(ctx, client, cmmsID) if fetchErr != nil { result.Failed++ result.Errors = append(result.Errors, fmt.Sprintf("%s: %v", cmmsID, fetchErr)) log.Printf("[%d/%d] FAILED %s: %v", i+1, result.Total, cmmsID, fetchErr) db.logAssetSync(ctx, cmmsID, customerID, "error", fetchErr.Error(), time.Since(assetStart).Milliseconds()) continue } tx, txErr := db.pool.Begin(ctx) if txErr != nil { result.Failed++ result.Errors = append(result.Errors, fmt.Sprintf("%s: tx begin: %v", cmmsID, txErr)) db.logAssetSync(ctx, cmmsID, customerID, "error", txErr.Error(), time.Since(assetStart).Milliseconds()) continue } enrichErr := db.EnrichAssetFromAPI(ctx, tx, apiResp) if enrichErr != nil { tx.Rollback(ctx) result.Failed++ result.Errors = append(result.Errors, fmt.Sprintf("%s: enrich: %v", cmmsID, enrichErr)) log.Printf("[%d/%d] ENRICH FAILED %s: %v", i+1, result.Total, cmmsID, enrichErr) db.logAssetSync(ctx, cmmsID, customerID, "error", enrichErr.Error(), time.Since(assetStart).Milliseconds()) continue } if commitErr := tx.Commit(ctx); commitErr != nil { result.Failed++ result.Errors = append(result.Errors, fmt.Sprintf("%s: commit: %v", cmmsID, commitErr)) db.logAssetSync(ctx, cmmsID, customerID, "error", commitErr.Error(), time.Since(assetStart).Milliseconds()) continue } result.Success++ db.logAssetSync(ctx, cmmsID, customerID, "success", "", time.Since(assetStart).Milliseconds()) log.Printf("[%d/%d] OK %s (%dms)", i+1, result.Total, cmmsID, time.Since(assetStart).Milliseconds()) if (i+1)%50 == 0 { elapsed := time.Since(start).Seconds() rate := float64(i+1) / elapsed log.Printf("Progress: %d/%d synced (%d failed, %.1f assets/sec, ~%.0f min remaining)", result.Success, result.Total, result.Failed, rate, float64(result.Total-i-1)/rate/60) } } db.pool.Exec(ctx, ` INSERT INTO cmms_sync_log (sync_type, status, assets_synced, duration_ms) VALUES ('api_sync', 'complete', $1, $2)`, result.Success, int(time.Since(start).Milliseconds())) log.Printf("Sync complete: %d success, %d failed, %d total in %dms", result.Success, result.Failed, result.Total, time.Since(start).Milliseconds()) result.Duration = strconv.FormatInt(time.Since(start).Milliseconds(), 10) return result, nil } func (db *DB) logAssetSync(ctx context.Context, cmmsID string, customerID int, status, errMsg string, durationMs int64) { var errPtr *string if errMsg != "" { errPtr = &errMsg } db.pool.Exec(ctx, ` INSERT INTO ic3_sync_log (serialnumber, sync_type, status, error_message, customer_id, duration_ms, synced_at) VALUES ($1, 'api_sync', $2, $3, $4, $5, NOW())`, cmmsID, status, errPtr, nullInt(customerID), int(durationMs)) } // getSyncAssetIDs returns cmms_asset_ids in report_seq_no order func (db *DB) getSyncAssetIDs(ctx context.Context, batchSize, offset, customerID int) ([]string, error) { limit := 9999 if batchSize > 0 { limit = batchSize } rows, err := db.pool.Query(ctx, ` SELECT am.cmms_asset_id FROM ic3_asset_master am WHERE am.cmms_asset_id IS NOT NULL AND ($3 = 0 OR am.customer_id = $3) ORDER BY am.report_seq_no ASC NULLS LAST, am.asset_id ASC LIMIT $1 OFFSET $2`, limit, offset, customerID) if err != nil { return nil, err } defer rows.Close() var ids []string for rows.Next() { var id string rows.Scan(&id) ids = append(ids, id) } return ids, nil } // getFailedSyncAssets returns cmms_asset_ids that had errors in last sync func (db *DB) getFailedSyncAssets(ctx context.Context, limit, customerID int) ([]string, error) { if limit <= 0 { limit = 9999 } rows, err := db.pool.Query(ctx, ` SELECT DISTINCT am.cmms_asset_id FROM ic3_asset_master am INNER JOIN ( SELECT serialnumber, MAX(synced_at) as last_sync FROM ic3_sync_log WHERE sync_type = 'api_sync' GROUP BY serialnumber ) latest ON latest.serialnumber = am.cmms_asset_id INNER JOIN ic3_sync_log sl ON sl.serialnumber = am.cmms_asset_id AND sl.synced_at = latest.last_sync AND sl.status = 'error' WHERE am.cmms_asset_id IS NOT NULL AND ($2 = 0 OR am.customer_id = $2) ORDER BY am.cmms_asset_id LIMIT $1`, limit, customerID) if err != nil { return nil, err } defer rows.Close() var ids []string for rows.Next() { var id string rows.Scan(&id) ids = append(ids, id) } return ids, nil } // ── helpers ────────────────────────────────────────────────────────────────── func nullStr(s string) any { if s == "" { return nil } return s } // nullStrFromInt converts int to string for TEXT columns, returns nil if 0 func nullStrFromInt(i int) any { if i == 0 { return nil } return strconv.Itoa(i) } // StartContinuousSyncWorker begins background syncing of all assets // Runs indefinitely, syncing batch by batch with configurable interval func (db *DB) StartContinuousSyncWorker(ctx context.Context, batchSize int, intervalSec int) { go func() { ticker := time.NewTicker(time.Duration(intervalSec) * time.Second) defer ticker.Stop() for { select { case <-ctx.Done(): log.Printf("Continuous sync worker stopped") return case <-ticker.C: // Get count of never-synced assets var neverSynced int db.pool.QueryRow(ctx, ` SELECT COUNT(*) FROM ic3_asset_master am WHERE NOT EXISTS ( SELECT 1 FROM ic3_sync_log sl WHERE sl.serialnumber = am.cmms_asset_id AND sl.sync_type = 'api_sync' )`).Scan(&neverSynced) if neverSynced == 0 { log.Printf("All assets synced! No more assets to process") continue } // Get current offset from last sync var lastOffset int db.pool.QueryRow(ctx, ` SELECT COALESCE(MAX(assets_synced), 0) FROM cmms_sync_log WHERE sync_type='api_sync' `).Scan(&lastOffset) log.Printf("Continuous sync: %d never-synced remaining, syncing next batch (offset %d)...", neverSynced, lastOffset) // Sync next batch syncCtx, cancel := context.WithTimeout(ctx, 30*time.Minute) result, err := db.SyncAllAssetsFromInnomaint(syncCtx, batchSize, lastOffset, false, innomaintCustomerId) cancel() if err != nil { log.Printf("Continuous sync error: %v", err) } else { log.Printf("Continuous sync batch: %d success, %d failed, %d total", result.Success, result.Failed, result.Total) } } } }() log.Printf("Continuous sync worker started (batch=%d, interval=%ds)", batchSize, intervalSec) } func boolToInt(b bool) int { if b { return 1 } return 0 }