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 }) }