FranklinWH-Toolbox/service/database/data_download_schema.go

package database

import (
	"fmt"

	"gorm.io/gorm"
)

const dataDownloadSchemaVersion int64 = 2

var dataDownloadSchemaV1 = []string{
	`CREATE TABLE IF NOT EXISTS data_download_content_state (
        document_id varchar(64) PRIMARY KEY,
        document_kind varchar(32) NOT NULL,
        content_generation bigint NOT NULL DEFAULT 0,
        schema_version varchar(32) NOT NULL DEFAULT 'v1',
        row_count bigint NOT NULL DEFAULT 0,
        checksum char(64) NOT NULL DEFAULT '',
        source_fetched_at timestamptz NOT NULL,
        updated_at timestamptz NOT NULL DEFAULT now()
    )`,
	`CREATE TABLE IF NOT EXISTS data_download_power_flow_meta (
        document_id varchar(64) PRIMARY KEY REFERENCES data_download_content_state(document_id) ON DELETE CASCADE,
        site_sn varchar(64) NOT NULL,
        local_date date NOT NULL,
        timezone varchar(80) NOT NULL
    )`,
	`CREATE TABLE IF NOT EXISTS data_download_power_flow_point (
        document_id varchar(64) NOT NULL REFERENCES data_download_content_state(document_id) ON DELETE CASCADE,
        point_index integer NOT NULL,
        sampled_at timestamptz,
        payload jsonb NOT NULL,
        PRIMARY KEY(document_id, point_index)
    )`,
	`CREATE TABLE IF NOT EXISTS data_download_mqtt_log_entry (
        document_id varchar(64) NOT NULL REFERENCES data_download_content_state(document_id) ON DELETE CASCADE,
        row_index integer NOT NULL,
        occurred_at timestamptz,
        subtype varchar(80) NOT NULL DEFAULT '',
        payload jsonb NOT NULL,
        PRIMARY KEY(document_id, row_index)
    )`,
	`CREATE TABLE IF NOT EXISTS data_download_event_alarm_entry (
        document_id varchar(64) NOT NULL REFERENCES data_download_content_state(document_id) ON DELETE CASCADE,
        row_index integer NOT NULL,
        occurred_at timestamptz,
        payload jsonb NOT NULL,
        PRIMARY KEY(document_id, row_index)
    )`,
	`CREATE TABLE IF NOT EXISTS data_download_update_log_entry (
        document_id varchar(64) NOT NULL REFERENCES data_download_content_state(document_id) ON DELETE CASCADE,
        row_index integer NOT NULL,
        device_sn varchar(64) NOT NULL DEFAULT '',
        device_type varchar(80) NOT NULL DEFAULT '',
        occurred_at timestamptz,
        payload jsonb NOT NULL,
        PRIMARY KEY(document_id, row_index)
    )`,
	`CREATE INDEX IF NOT EXISTS idx_dd_pf_site_date ON data_download_power_flow_meta(site_sn, local_date)`,
	`CREATE INDEX IF NOT EXISTS idx_dd_mqtt_occurred ON data_download_mqtt_log_entry(document_id, occurred_at)`,
	`CREATE INDEX IF NOT EXISTS idx_dd_event_occurred ON data_download_event_alarm_entry(document_id, occurred_at)`,
	`CREATE INDEX IF NOT EXISTS idx_dd_update_device ON data_download_update_log_entry(document_id, device_sn, device_type)`,
	`ALTER TABLE data_download_content_state ADD COLUMN IF NOT EXISTS content_checksum char(64) NOT NULL DEFAULT ''`,
	`ALTER TABLE data_download_content_state ALTER COLUMN schema_version DROP DEFAULT`,
	`ALTER TABLE data_download_content_state ALTER COLUMN schema_version TYPE varchar(32) USING schema_version::text`,
	`ALTER TABLE data_download_content_state ALTER COLUMN schema_version SET DEFAULT 'v1'`,
	`DO $$ BEGIN IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'chk_dd_document_kind') THEN ALTER TABLE data_download_content_state ADD CONSTRAINT chk_dd_document_kind CHECK (document_kind IN ('power_flow','mqtt_log','event_alarm','update_log')); END IF; END $$`,
	`ALTER TABLE data_download_power_flow_meta ADD COLUMN IF NOT EXISTS reported_index integer NOT NULL DEFAULT 0`,
	`ALTER TABLE data_download_power_flow_point ADD COLUMN IF NOT EXISTS seq integer`,
	`ALTER TABLE data_download_power_flow_point ADD COLUMN IF NOT EXISTS device_time_local timestamp without time zone`,
	`ALTER TABLE data_download_power_flow_point ADD COLUMN IF NOT EXISTS device_time_utc timestamptz`,
	`ALTER TABLE data_download_power_flow_point ADD COLUMN IF NOT EXISTS power_fhp real`,
	`ALTER TABLE data_download_power_flow_point ADD COLUMN IF NOT EXISTS power_gen real`,
	`ALTER TABLE data_download_power_flow_point ADD COLUMN IF NOT EXISTS power_load real`,
	`ALTER TABLE data_download_power_flow_point ADD COLUMN IF NOT EXISTS power_sun real`,
	`ALTER TABLE data_download_power_flow_point ADD COLUMN IF NOT EXISTS power_uti real`,
	`ALTER TABLE data_download_power_flow_point ADD COLUMN IF NOT EXISTS soc real`,
	`ALTER TABLE data_download_power_flow_point ADD COLUMN IF NOT EXISTS kwh_total real`,
	`ALTER TABLE data_download_power_flow_point ADD COLUMN IF NOT EXISTS run_status integer`,
	`ALTER TABLE data_download_mqtt_log_entry ADD COLUMN IF NOT EXISTS cmd_type varchar(32)`,
	`ALTER TABLE data_download_mqtt_log_entry ADD COLUMN IF NOT EXISTS cmd_name varchar(255)`,
	`ALTER TABLE data_download_mqtt_log_entry ADD COLUMN IF NOT EXISTS gateway_id varchar(64)`,
	`ALTER TABLE data_download_mqtt_log_entry ADD COLUMN IF NOT EXISTS gateway_type varchar(80)`,
	`ALTER TABLE data_download_mqtt_log_entry ADD COLUMN IF NOT EXISTS gateway_status varchar(80)`,
	`ALTER TABLE data_download_mqtt_log_entry ADD COLUMN IF NOT EXISTS active_status varchar(80)`,
	`ALTER TABLE data_download_mqtt_log_entry ADD COLUMN IF NOT EXISTS device_time_raw text`,
	`ALTER TABLE data_download_mqtt_log_entry ADD COLUMN IF NOT EXISTS event_time_raw text`,
	`ALTER TABLE data_download_mqtt_log_entry ADD COLUMN IF NOT EXISTS device_time_local timestamp without time zone`,
	`ALTER TABLE data_download_mqtt_log_entry ADD COLUMN IF NOT EXISTS event_time_utc timestamptz`,
	`ALTER TABLE data_download_mqtt_log_entry ADD COLUMN IF NOT EXISTS log_data_raw text`,
	`ALTER TABLE data_download_mqtt_log_entry ADD COLUMN IF NOT EXISTS log_data_json jsonb`,
	`ALTER TABLE data_download_event_alarm_entry ADD COLUMN IF NOT EXISTS source_id varchar(80)`,
	`ALTER TABLE data_download_event_alarm_entry ADD COLUMN IF NOT EXISTS cmd_type varchar(80)`,
	`ALTER TABLE data_download_event_alarm_entry ADD COLUMN IF NOT EXISTS cmd_name varchar(255)`,
	`ALTER TABLE data_download_event_alarm_entry ADD COLUMN IF NOT EXISTS gateway_id varchar(64)`,
	`ALTER TABLE data_download_event_alarm_entry ADD COLUMN IF NOT EXISTS log_type varchar(32)`,
	`ALTER TABLE data_download_event_alarm_entry ADD COLUMN IF NOT EXISTS alarm_code varchar(80)`,
	`ALTER TABLE data_download_event_alarm_entry ADD COLUMN IF NOT EXISTS log_name text`,
	`ALTER TABLE data_download_event_alarm_entry ADD COLUMN IF NOT EXISTS log_data_raw text`,
	`ALTER TABLE data_download_event_alarm_entry ADD COLUMN IF NOT EXISTS log_data_json jsonb`,
	`ALTER TABLE data_download_event_alarm_entry ADD COLUMN IF NOT EXISTS device_time_raw text`,
	`ALTER TABLE data_download_event_alarm_entry ADD COLUMN IF NOT EXISTS start_time_raw text`,
	`ALTER TABLE data_download_event_alarm_entry ADD COLUMN IF NOT EXISTS end_time_raw text`,
	`CREATE TABLE IF NOT EXISTS data_download_update_order (
        document_id varchar(64) NOT NULL REFERENCES data_download_content_state(document_id) ON DELETE CASCADE,
        order_no varchar(128) NOT NULL, payload jsonb NOT NULL,
        PRIMARY KEY(document_id, order_no)
    )`,
	`CREATE TABLE IF NOT EXISTS data_download_update_device_version (
        document_id varchar(64) NOT NULL REFERENCES data_download_content_state(document_id) ON DELETE CASCADE,
        order_no varchar(128) NOT NULL, scope varchar(32) NOT NULL, device_sn varchar(64) NOT NULL,
        name varchar(255) NOT NULL, payload jsonb NOT NULL,
        PRIMARY KEY(document_id, order_no, scope, device_sn, name)
    )`,
	`CREATE TABLE IF NOT EXISTS data_download_ahub_update (
        document_id varchar(64) NOT NULL REFERENCES data_download_content_state(document_id) ON DELETE CASCADE,
        order_no varchar(128) NOT NULL, ahub_sn varchar(64) NOT NULL, payload jsonb NOT NULL,
        PRIMARY KEY(document_id, order_no, ahub_sn)
    )`,
	`CREATE TABLE IF NOT EXISTS data_download_ahub_update_detail (
        document_id varchar(64) NOT NULL REFERENCES data_download_content_state(document_id) ON DELETE CASCADE,
        order_no varchar(128) NOT NULL, ahub_sn varchar(64) NOT NULL, file_order varchar(128) NOT NULL,
        file_name varchar(255) NOT NULL, payload jsonb NOT NULL,
        PRIMARY KEY(document_id, order_no, ahub_sn, file_order, file_name)
    )`,
	`CREATE INDEX IF NOT EXISTS idx_dd_mqtt_local ON data_download_mqtt_log_entry(document_id, device_time_local, row_index)`,
	`CREATE INDEX IF NOT EXISTS idx_dd_event_alarm ON data_download_event_alarm_entry(document_id, log_type, alarm_code)`,
}

func MigrateDataDownloadSchema(db *gorm.DB) error {
	if db == nil {
		return fmt.Errorf("PostgreSQL is not initialized")
	}
	return db.Transaction(func(tx *gorm.DB) error {
		if err := tx.Exec(`CREATE TABLE IF NOT EXISTS data_download_schema_migrations (
            version bigint PRIMARY KEY,
            applied_at timestamptz NOT NULL DEFAULT now()
        )`).Error; err != nil {
			return err
		}
		var count int64
		if err := tx.Raw("SELECT count(*) FROM data_download_schema_migrations WHERE version = ?", dataDownloadSchemaVersion).Scan(&count).Error; err != nil {
			return err
		}
		if count != 0 {
			return nil
		}
		for _, statement := range dataDownloadSchemaV1 {
			if err := tx.Exec(statement).Error; err != nil {
				return fmt.Errorf("apply data-download schema v%d: %w", dataDownloadSchemaVersion, err)
			}
		}
		return tx.Exec("INSERT INTO data_download_schema_migrations(version) VALUES (?)", dataDownloadSchemaVersion).Error
	})
}