FranklinWH-Toolbox/service/repositories/pgsql/dataDownloadRepo.go

package pgsql

import (
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"math"
	"strconv"
	"strings"
	"time"

	"fwh-toolbox-server/database"
	"fwh-toolbox-server/domain"
	"fwh-toolbox-server/dto"

	"gorm.io/gorm"
)

type DataDownloadRow struct {
	OccurredAt *time.Time
	Subtype    string
	DeviceSN   string
	DeviceType string
	Payload    any
}

type powerFlowPoint struct {
	Sequence int `json:"sequence"`
	// DeviceTimeLocal preserves the wall-clock value reported by the device;
	// DataDownloadRow.OccurredAt stores the same instant normalized to UTC.
	DeviceTimeLocal                                                  string `json:"deviceTimeLocal"`
	PowerFhp, PowerGen, PowerLoad, PowerSun, PowerUti, Soc, KwhTotal *float32
	RunStatus                                                        *int32 `json:"runStatus"`
}

type DataDownloadContentState struct {
	DocumentID        string    `gorm:"column:document_id" json:"documentId"`
	DocumentKind      string    `gorm:"column:document_kind" json:"documentKind"`
	ContentGeneration int64     `gorm:"column:content_generation" json:"contentGeneration"`
	SchemaVersion     string    `gorm:"column:schema_version" json:"schemaVersion"`
	RowCount          int64     `gorm:"column:row_count" json:"rowCount"`
	Checksum          string    `gorm:"column:checksum" json:"checksum"`
	SourceFetchedAt   time.Time `gorm:"column:source_fetched_at" json:"sourceFetchedAt"`
	UpdatedAt         time.Time `gorm:"column:updated_at" json:"updatedAt"`
}

func ReplaceDataDownloadRows(documentID string, kind domain.DataDownloadKind, siteSN, localDate, timezone string, rows []DataDownloadRow, fetchedAt time.Time) (*DataDownloadContentState, error) {
	var state DataDownloadContentState
	reportedIndex := uint(0)
	if kind == domain.DataDownloadPowerFlow && len(rows) == 1 {
		if value, ok := rows[0].Payload.(*dto.SitePowerFlow); ok {
			reportedIndex = value.Index
			points, err := expandPowerFlow(value, timezone)
			if err != nil {
				return nil, err
			}
			rows = points
		}
	}
	err := database.PgSQL.Transaction(func(tx *gorm.DB) error {
		// The transaction-scoped advisory lock makes replacement and readers mutually exclusive.
		if err := tx.Exec("SELECT pg_advisory_xact_lock(hashtextextended(?, 0))", documentID).Error; err != nil {
			return err
		}
		var generation int64
		if err := tx.Raw("SELECT content_generation FROM data_download_content_state WHERE document_id = ? FOR UPDATE", documentID).Scan(&generation).Error; err != nil {
			return err
		}
		generation++
		payloadBytes, err := json.Marshal(rows)
		if err != nil {
			return err
		}
		sum := sha256.Sum256(payloadBytes)
		checksum := hex.EncodeToString(sum[:])
		if err := tx.Exec(`INSERT INTO data_download_content_state
            (document_id, document_kind, content_generation, schema_version, row_count, checksum, content_checksum, source_fetched_at, updated_at)
            VALUES (?, ?, ?, 'v1', ?, ?, ?, ?, now())
            ON CONFLICT (document_id) DO UPDATE SET document_kind=excluded.document_kind,
              content_generation=excluded.content_generation, schema_version=excluded.schema_version,
              row_count=excluded.row_count, checksum=excluded.checksum, content_checksum=excluded.content_checksum,
              source_fetched_at=excluded.source_fetched_at, updated_at=now()`,
			documentID, string(kind), generation, len(rows), checksum, checksum, fetchedAt).Error; err != nil {
			return err
		}

		table := ""
		switch kind {
		case domain.DataDownloadPowerFlow:
			table = "data_download_power_flow_point"
			if err := tx.Exec("DELETE FROM data_download_power_flow_meta WHERE document_id = ?", documentID).Error; err != nil {
				return err
			}
			if err := tx.Exec("INSERT INTO data_download_power_flow_meta(document_id, site_sn, local_date, timezone, reported_index) VALUES (?, ?, ?, ?, ?)", documentID, siteSN, localDate, timezone, reportedIndex).Error; err != nil {
				return err
			}
		case domain.DataDownloadMqttLog:
			table = "data_download_mqtt_log_entry"
		case domain.DataDownloadEventAlarm:
			table = "data_download_event_alarm_entry"
		case domain.DataDownloadUpdateLog:
			for _, name := range []string{"data_download_update_log_entry", "data_download_update_order", "data_download_update_device_version", "data_download_ahub_update", "data_download_ahub_update_detail"} {
				if err := tx.Exec("DELETE FROM "+name+" WHERE document_id = ?", documentID).Error; err != nil {
					return err
				}
			}
		default:
			return fmt.Errorf("%s is not a structured document kind", kind)
		}
		if table != "" {
			if err := tx.Exec("DELETE FROM "+table+" WHERE document_id = ?", documentID).Error; err != nil {
				return err
			}
		}
		for i, row := range rows {
			payload, err := json.Marshal(row.Payload)
			if err != nil {
				return err
			}
			switch kind {
			case domain.DataDownloadPowerFlow:
				point := row.Payload.(powerFlowPoint)
				err = tx.Exec(`INSERT INTO data_download_power_flow_point(document_id, point_index, seq, sampled_at, device_time_local, device_time_utc, power_fhp, power_gen, power_load, power_sun, power_uti, soc, kwh_total, run_status, payload)
                    VALUES (?, ?, ?, ?, NULLIF(?, '')::timestamp, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?::jsonb)`, documentID, i, point.Sequence, row.OccurredAt, point.DeviceTimeLocal, row.OccurredAt, point.PowerFhp, point.PowerGen, point.PowerLoad, point.PowerSun, point.PowerUti, point.Soc, point.KwhTotal, point.RunStatus, string(payload)).Error
			case domain.DataDownloadMqttLog:
				value := row.Payload.(domain.MqttLogItem)
				var logJSON any = nil
				if json.Valid([]byte(value.LogData)) {
					logJSON = value.LogData
				}
				deviceLocal := parseLocalTime(value.LogTime)
				err = tx.Exec(`INSERT INTO data_download_mqtt_log_entry(document_id,row_index,occurred_at,subtype,cmd_type,cmd_name,gateway_id,gateway_type,gateway_status,active_status,device_time_raw,event_time_raw,device_time_local,event_time_utc,log_data_raw,log_data_json,payload)
                    VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?::jsonb,?::jsonb)`, documentID, i, row.OccurredAt, row.Subtype, value.CmdType, value.CmdName, value.GatewayId, value.GatewayType, value.GatewayStatus, value.ActiveStatus, value.LogTime, value.Time, deviceLocal, row.OccurredAt, value.LogData, logJSON, string(payload)).Error
			case domain.DataDownloadEventAlarm:
				value := row.Payload.(domain.EventAlarmItem)
				var logJSON any = nil
				if json.Valid([]byte(value.LogData)) {
					logJSON = value.LogData
				}
				sourceID := ""
				if value.ID != nil {
					sourceID = *value.ID
				}
				endRaw := ""
				if value.EndTime != nil {
					endRaw = *value.EndTime
				}
				err = tx.Exec(`INSERT INTO data_download_event_alarm_entry(document_id,row_index,occurred_at,source_id,cmd_type,cmd_name,gateway_id,log_type,alarm_code,log_name,log_data_raw,log_data_json,device_time_raw,start_time_raw,end_time_raw,payload)
                    VALUES (?,?,?,?,?,?,?,?,?,?,?,?::jsonb,?,?,?,?::jsonb)`, documentID, i, row.OccurredAt, sourceID, value.CmdType, value.CmdName, value.GatewayId, value.LogType, value.AlarmCode, value.LogName, value.LogData, logJSON, value.LogTime, value.StartTime, endRaw, string(payload)).Error
			case domain.DataDownloadUpdateLog:
				switch value := row.Payload.(type) {
				case domain.SystemUpdateRecordInfo:
					err = tx.Exec("INSERT INTO data_download_update_order(document_id,order_no,payload) VALUES (?,?,?::jsonb)", documentID, value.OrderNo, string(payload)).Error
				case domain.SystemUpdateDetailInfo:
					err = tx.Exec("INSERT INTO data_download_update_device_version(document_id,order_no,scope,device_sn,name,payload) VALUES (?,?,?,?,?,?::jsonb)", documentID, value.OrderNo, "gateway", value.GatewayId, strconv.FormatInt(value.Id, 10), string(payload)).Error
				case domain.AHubUpdateRecordInfo:
					err = tx.Exec("INSERT INTO data_download_ahub_update(document_id,order_no,ahub_sn,payload) VALUES (?,?,?,?::jsonb)", documentID, value.OrderNo, value.AHubSn, string(payload)).Error
					for _, detail := range value.DetailList {
						detailPayload, _ := json.Marshal(detail)
						if insertErr := tx.Exec("INSERT INTO data_download_ahub_update_detail(document_id,order_no,ahub_sn,file_order,file_name,payload) VALUES (?,?,?,?,?,?::jsonb)", documentID, value.OrderNo, value.AHubSn, detail.UpgradeFileOrder, detail.UpgradeFileName, string(detailPayload)).Error; insertErr != nil {
							return insertErr
						}
					}
				}
			}
			if err != nil {
				return err
			}
		}
		return tx.Raw("SELECT * FROM data_download_content_state WHERE document_id = ?", documentID).Scan(&state).Error
	})
	return &state, err
}

func expandPowerFlow(value *dto.SitePowerFlow, timezone string) ([]DataDownloadRow, error) {
	length := len(value.DeviceTimeArray)
	for name, size := range map[string]int{"powerFhp": len(value.PowerFhpArray), "powerGen": len(value.PowerGenArray), "powerLoad": len(value.PowerLoadArray), "powerSun": len(value.PowerSunArray), "powerUti": len(value.PowerUtiArray), "soc": len(value.SocArray), "runStatus": len(value.RunStatusArray), "kwhTotal": len(value.KwhTotalArray)} {
		if size > length {
			return nil, fmt.Errorf("PowerFlow %s has %d values but only %d timestamps", name, size, length)
		}
	}
	for _, status := range value.RunStatusArray {
		if uint64(status) > uint64(math.MaxInt32) {
			return nil, fmt.Errorf("PowerFlow runStatus %d exceeds PostgreSQL integer range", status)
		}
	}
	rows := make([]DataDownloadRow, 0, length)
	for i, raw := range value.DeviceTimeArray {
		point := powerFlowPoint{Sequence: i, DeviceTimeLocal: raw, PowerFhp: f32At(value.PowerFhpArray, i), PowerGen: f32At(value.PowerGenArray, i), PowerLoad: f32At(value.PowerLoadArray, i), PowerSun: f32At(value.PowerSunArray, i), PowerUti: f32At(value.PowerUtiArray, i), Soc: f32At(value.SocArray, i), KwhTotal: f32At(value.KwhTotalArray, i), RunStatus: int32At(value.RunStatusArray, i)}
		var occurred *time.Time
		loc := postgresDataDownloadLocation(timezone)
		if parsed, parseErr := time.ParseInLocation(time.DateTime, raw, loc); parseErr == nil {
			utc := parsed.UTC()
			occurred = &utc
		}
		rows = append(rows, DataDownloadRow{OccurredAt: occurred, Payload: point})
	}
	return rows, nil
}

func parseLocalTime(value string) *time.Time {
	parsed, err := time.Parse(time.DateTime, value)
	if err != nil {
		return nil
	}
	return &parsed
}
func postgresDataDownloadLocation(name string) *time.Location {
	if loc, err := time.LoadLocation(name); err == nil {
		return loc
	}
	if strings.HasPrefix(name, "UTC") {
		if hours, err := strconv.ParseFloat(strings.TrimPrefix(name, "UTC"), 64); err == nil {
			return time.FixedZone(name, int(math.Round(hours*3600)))
		}
	}
	return time.UTC
}
func f32At(values []float32, index int) *float32 {
	if index >= len(values) {
		return nil
	}
	value := values[index]
	return &value
}
func int32At(values []uint, index int) *int32 {
	if index >= len(values) {
		return nil
	}
	value := int32(values[index])
	return &value
}

func ReadDataDownloadRows(documentID string, kind domain.DataDownloadKind) (*DataDownloadContentState, []json.RawMessage, error) {
	var state DataDownloadContentState
	var rows []json.RawMessage
	err := database.PgSQL.Transaction(func(tx *gorm.DB) error {
		if err := tx.Exec("SELECT pg_advisory_xact_lock_shared(hashtextextended(?, 0))", documentID).Error; err != nil {
			return err
		}
		if err := tx.Raw("SELECT * FROM data_download_content_state WHERE document_id = ?", documentID).Scan(&state).Error; err != nil {
			return err
		}
		table := map[domain.DataDownloadKind]string{
			domain.DataDownloadPowerFlow:  "data_download_power_flow_point",
			domain.DataDownloadMqttLog:    "data_download_mqtt_log_entry",
			domain.DataDownloadEventAlarm: "data_download_event_alarm_entry",
			domain.DataDownloadUpdateLog:  "",
		}[kind]
		if table == "" && kind != domain.DataDownloadUpdateLog {
			return fmt.Errorf("%s is not structured", kind)
		}
		type payloadRow struct {
			Payload []byte `gorm:"column:payload"`
		}
		var values []payloadRow
		if kind == domain.DataDownloadUpdateLog {
			if err := tx.Raw(`SELECT payload FROM (
                    SELECT 1 AS section, order_no AS k1, '' AS k2, payload FROM data_download_update_order WHERE document_id = ?
                    UNION ALL SELECT 2, order_no, device_sn, payload FROM data_download_update_device_version WHERE document_id = ?
                    UNION ALL SELECT 3, order_no, ahub_sn, payload FROM data_download_ahub_update WHERE document_id = ?
                    UNION ALL SELECT 4, order_no, ahub_sn || file_order || file_name, payload FROM data_download_ahub_update_detail WHERE document_id = ?
                ) current_update ORDER BY section, k1, k2`, documentID, documentID, documentID, documentID).Scan(&values).Error; err != nil {
				return err
			}
			for _, value := range values {
				rows = append(rows, json.RawMessage(value.Payload))
			}
			return nil
		}
		order := "row_index"
		if kind == domain.DataDownloadPowerFlow {
			order = "point_index"
		}
		if err := tx.Raw("SELECT payload FROM "+table+" WHERE document_id = ? ORDER BY "+order, documentID).Scan(&values).Error; err != nil {
			return err
		}
		for _, value := range values {
			rows = append(rows, json.RawMessage(value.Payload))
		}
		return nil
	})
	return &state, rows, err
}

func DeleteDataDownloadContent(documentID string) error {
	return database.PgSQL.Transaction(func(tx *gorm.DB) error {
		if err := tx.Exec("SELECT pg_advisory_xact_lock(hashtextextended(?, 0))", documentID).Error; err != nil {
			return err
		}
		return tx.Exec("DELETE FROM data_download_content_state WHERE document_id = ?", documentID).Error
	})
}