package migrate

import (
	"github.com/pingcap/log"
	"go.uber.org/zap"
	"gorm.io/gorm"
)

var _ = AddDBMigrateRecord(
	"202506031500",
	func(db *gorm.DB) error {
		cdcChangefeeds := `
  CREATE TABLE IF NOT EXISTS cdc_changefeeds (
  id VARCHAR(255) NOT NULL,
  changefeed_id VARCHAR(255) NOT NULL,
  tenant_id VARCHAR(255) NOT NULL,
  project_id VARCHAR(255) NOT NULL,
  cluster_id VARCHAR(255) NOT NULL,
  keyspace_id BIGINT NOT NULL,
  cloud_provider VARCHAR(255) NOT NULL,
  region VARCHAR(255) NOT NULL,
  kubernetes VARCHAR(255) NOT NULL,
  state VARCHAR(32) NOT NULL,
  name VARCHAR(255) NOT NULL,
  sink_type VARCHAR(32) NOT NULL,
  kafka JSON DEFAULT NULL,
  mysql JSON DEFAULT NULL,
  filter JSON DEFAULT NULL,
  start_mode VARCHAR(32),
  start_tso BIGINT DEFAULT NULL,
  target_tso BIGINT DEFAULT NULL,
  checkpoint_tso BIGINT DEFAULT NULL,
  error TEXT,
  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at TIMESTAMP NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
  deleted_at TIMESTAMP NULL DEFAULT NULL,
  running_at TIMESTAMP NULL DEFAULT NULL,
  last_sync_at TIMESTAMP NULL DEFAULT NULL,
  PRIMARY KEY (id),
  INDEX idx_cluster_id (cluster_id),
  INDEX idx_changefeed_id (changefeed_id),
  INDEX idx_created_at (created_at),
  INDEX idx_state (state)
);
`
		err := db.Exec(cdcChangefeeds).Error
		if err != nil {
			log.Error("create table cdc_changefeeds failed", zap.Error(err))
		}

		cdcCluster := `
  CREATE TABLE IF NOT EXISTS cdc_cluster (
  cluster_id VARCHAR(255) NOT NULL,
  tenant_id VARCHAR(255) NOT NULL,
  project_id VARCHAR(255) NOT NULL,
  cloud_provider VARCHAR(255) NOT NULL,
  region VARCHAR(255) NOT NULL,
  kubernetes VARCHAR(255) NOT NULL,
  keyspace_id BIGINT NOT NULL,
  cluster_status_id BIGINT NOT NULL,
  changefeed_count BIGINT NOT NULL DEFAULT 1,
  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at TIMESTAMP NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
  state VARCHAR(32) NOT NULL,
  PRIMARY KEY (cluster_id),
  INDEX idx_count (changefeed_count),
  INDEX idx_state (state)
);
`
		err = db.Exec(cdcCluster).Error
		if err != nil {
			log.Error("create table cdc_cluster failed", zap.Error(err))
		}

		privateLinkEndpoint := `
  CREATE TABLE IF NOT EXISTS cdc_pl_endpoint (
  id VARCHAR(255) NOT NULL,
  tenant_id VARCHAR(255) NOT NULL,
  project_id VARCHAR(255) NOT NULL,
  cluster_id VARCHAR(255) NOT NULL,
  cloud_provider VARCHAR(255) NOT NULL,
  region VARCHAR(255) NOT NULL,
  kubernetes VARCHAR(255) NOT NULL,
  service_name VARCHAR(255) NOT NULL,
  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at TIMESTAMP NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
  sink_type VARCHAR(32) NOT NULL,
  state VARCHAR(32) NOT NULL,
  message TEXT DEFAULT NULL,
  kafka_type VARCHAR(32) DEFAULT NULL,
  kafka_id VARCHAR(255) DEFAULT NULL,
  advertised_listeners_pattern JSON DEFAULT NULL,
  endpoint_id VARCHAR(255) DEFAULT NULL,
  dns_configured BOOLEAN NOT NULL DEFAULT FALSE,
  
  PRIMARY KEY (id),
  UNIQUE KEY udx_service_name (service_name),
  UNIQUE KEY udx_kafka (kafka_type, kafka_id),
  INDEX idx_cluster_id (cluster_id),
  INDEX idx_state (state)
);`

		err = db.Exec(privateLinkEndpoint).Error
		if err != nil {
			log.Error("create table cdc_pl_endpoint failed", zap.Error(err))
		}

		return nil
	},
)