package configs

import (
	"fmt"
	"time"
)

type Syncers struct {
	AzBlob AzBlobSyncer `toml:"azblob"`
	Gcs    GcsSyncer    `toml:"gcs"`
	Oss    OssSyncer    `toml:"oss"`
	S3     S3Syncer     `toml:"s3"`
}

func (r *Syncers) Validate(mode string) error {
	if r.AzBlob.Enabled {
		return r.AzBlob.Validate(mode)
	}
	if r.Gcs.Enabled {
		return r.Gcs.Validate(mode)
	}
	if r.Oss.Enabled {
		return r.Oss.Validate(mode)
	}
	if r.S3.Enabled {
		return r.S3.Validate(mode)
	}
	return fmt.Errorf("no syncers enabled")
}

func (r *Syncers) Shards(mode string) []FileShard {
	if r.AzBlob.Enabled {
		return r.AzBlob.Shards(mode)
	}
	if r.Gcs.Enabled {
		return r.Gcs.Shards(mode)
	}
	if r.Oss.Enabled {
		return r.Oss.Shards(mode)
	}
	if r.S3.Enabled {
		return r.S3.Shards(mode)
	}
	return nil
}

// CommonSyncer holds the shared configuration fields for all storage-provider syncers.
type CommonSyncer struct {
	Enabled    bool     `toml:"enabled"`
	StartTime  int64    `toml:"startTime"`
	EndTime    int64    `toml:"endTime"`
	OrgID      string   `toml:"orgId"`
	ClusterID  string   `toml:"clusterId"`
	Components []string `toml:"components"`
}

func (c *CommonSyncer) validateCommon(provider string) error {
	if c.ClusterID == "" {
		return fmt.Errorf("syncers.%s.clusterId is required", provider)
	}
	if len(c.Components) == 0 {
		return fmt.Errorf("syncers.%s.components is required", provider)
	}
	if c.StartTime <= 0 {
		return fmt.Errorf("syncers.%s.startTime must be > 0", provider)
	}
	if c.EndTime <= c.StartTime {
		return fmt.Errorf("syncers.%s.endTime must be > startTime", provider)
	}
	return nil
}

func (c *CommonSyncer) toShards(mode, destPrefix string) []FileShard {
	switch mode {
	case ModeLog:
		hours := generateHours(c.StartTime, c.EndTime)
		return generateLogShards(hours, c.Components, c.ClusterID, destPrefix)
	case ModeDeltalake:
		dates := generateDates(c.StartTime, c.EndTime)
		return generateDeltalakeShards(dates, c.Components, c.OrgID, c.ClusterID, destPrefix)
	}
	return nil
}

func (c *CommonSyncer) validateMode(mode string) error {
	if mode == ModeDeltalake && c.OrgID == "" {
		return fmt.Errorf("orgId is required for deltalake mode")
	}
	return nil
}

type S3Syncer struct {
	CommonSyncer `toml:",squash"`
	Source       S3Sink `toml:"source"`
	Dest         S3Sink `toml:"dest"`
}

func (s *S3Syncer) Validate(mode string) error {
	if err := s.validateCommon("s3"); err != nil {
		return err
	}
	if err := s.validateMode(mode); err != nil {
		return err
	}
	if err := s.Source.Validate(); err != nil {
		return err
	}
	if err := s.Dest.Validate(); err != nil {
		return err
	}
	return nil
}

func (s *S3Syncer) Shards(mode string) []FileShard {
	return s.toShards(mode, s.Dest.PathPrefix)
}

type AzBlobSyncer struct {
	CommonSyncer `toml:",squash"`
	Source       AzBlobSink `toml:"source"`
	Dest         AzBlobSink `toml:"dest"`
}

func (s *AzBlobSyncer) Validate(mode string) error {
	if err := s.validateCommon("azblob"); err != nil {
		return err
	}
	if err := s.validateMode(mode); err != nil {
		return err
	}
	if err := s.Source.Validate(); err != nil {
		return err
	}
	if err := s.Dest.Validate(); err != nil {
		return err
	}
	return nil
}

func (s *AzBlobSyncer) Shards(mode string) []FileShard {
	return s.toShards(mode, s.Dest.PathPrefix)
}

type GcsSyncer struct {
	CommonSyncer `toml:",squash"`
	Source       GcsSink `toml:"source"`
	Dest         GcsSink `toml:"dest"`
}

func (s *GcsSyncer) Validate(mode string) error {
	if err := s.validateCommon("gcs"); err != nil {
		return err
	}
	if err := s.validateMode(mode); err != nil {
		return err
	}
	if err := s.Source.Validate(); err != nil {
		return err
	}
	if err := s.Dest.Validate(); err != nil {
		return err
	}
	return nil
}

func (s *GcsSyncer) Shards(mode string) []FileShard {
	return s.toShards(mode, s.Dest.PathPrefix)
}

type OssSyncer struct {
	CommonSyncer `toml:",squash"`
	Source       OssSink `toml:"source"`
	Dest         OssSink `toml:"dest"`
}

func (s *OssSyncer) Validate(mode string) error {
	if err := s.validateCommon("oss"); err != nil {
		return err
	}
	if err := s.validateMode(mode); err != nil {
		return err
	}
	if err := s.Source.Validate(); err != nil {
		return err
	}
	if err := s.Dest.Validate(); err != nil {
		return err
	}
	return nil
}

func (s *OssSyncer) Shards(mode string) []FileShard {
	return s.toShards(mode, s.Dest.PathPrefix)
}

// generateHours produces hourly bucket strings for a time range.
//
// Hourly granularity matches how log pipelines rotate files on hour boundaries,
// so each hour corresponds to a distinct set of log objects to replicate.
func generateHours(start, end int64) []string {
	s := time.Unix(start, 0).UTC().Truncate(time.Hour)
	e := time.Unix(end, 0).UTC().Truncate(time.Hour)

	var hours []string
	for t := s; !t.After(e); t = t.Add(time.Hour) {
		hours = append(hours, t.Format("2006010215"))
	}
	return hours
}

// generateDates produces daily bucket strings for a time range.
//
// Daily granularity matches how deltalake pipelines partition data by date,
// so each date corresponds to a distinct set of deltalake objects to replicate.
func generateDates(start, end int64) []string {
	s := time.Unix(start, 0).UTC().Truncate(24 * time.Hour)
	e := time.Unix(end, 0).UTC().Truncate(24 * time.Hour)

	var dates []string
	for t := s; !t.After(e); t = t.AddDate(0, 0, 1) {
		dates = append(dates, t.Format("20060102"))
	}
	return dates
}

// generateLogShards creates one shard per (hour, component) pair.
//
// Source prefix: diagnosis/data/{clusterId}/merged-logs/{hour}/{component}
// Dest prefix: {destPrefix}/{clusterId}/{hour}/{component}
func generateLogShards(hours, components []string, clusterID, destPrefix string) []FileShard {
	var index int
	var shards []FileShard
	for _, hour := range hours {
		for _, comp := range components {
			shards = append(shards, FileShard{
				Index:        index,
				SourcePrefix: fmt.Sprintf("diagnosis/data/%s/merged-logs/%s/%s", clusterID, hour, comp),
				DestPrefix:   fmt.Sprintf("%s/%s/%s/%s", destPrefix, clusterID, hour, comp),
				Hour:         hour,
				Component:    comp,
				ClusterID:    clusterID,
			})
			index++
		}
	}
	return shards
}

// generateDeltalakeShards creates one shard per (date, component) pair.
//
// Source prefix: deltalake/{orgId}/{clusterId}/{component}/date={date}
// Dest prefix: {destPrefix}/{orgId}/{clusterId}/{component}/date={date}
func generateDeltalakeShards(dates, components []string, orgID, clusterID, destPrefix string) []FileShard {
	var index int
	var shards []FileShard
	for _, date := range dates {
		for _, comp := range components {
			shards = append(shards, FileShard{
				Index:        index,
				SourcePrefix: fmt.Sprintf("deltalake/%s/%s/%s/date=%s", orgID, clusterID, comp, date),
				DestPrefix:   fmt.Sprintf("%s/%s/%s/%s/date=%s", destPrefix, orgID, clusterID, comp, date),
				Date:         date,
				Component:    comp,
				ClusterID:    clusterID,
			})
			index++
		}
	}
	return shards
}