package process

import (
	"bufio"
	"compress/gzip"
	"context"
	"database/sql"
	"fmt"
	"io"
	"os"
	"path"
	"path/filepath"
	"sort"
	"strings"
	"time"

	"github.com/pingcap/log"
	"github.com/tidbcloud/observability/cmd/diagnosis-compute/pkg/config"
	"github.com/tidbcloud/observability/cmd/diagnosis-compute/pkg/deltalake"
	"github.com/tidbcloud/observability/cmd/diagnosis-compute/pkg/logutil"
	"github.com/tidbcloud/observability/cmd/diagnosis-compute/pkg/metrics"
	parquetpkg "github.com/tidbcloud/observability/cmd/diagnosis-compute/pkg/parquet"
	"github.com/tidbcloud/observability/cmd/diagnosis-compute/pkg/types"
	"github.com/tidbcloud/observability/o11ydiagnosis/pkg/storage"
	"go.uber.org/zap"
)

const hourSeconds int64 = 3600

const hourProcessingDelay = 30 * time.Minute

// isHourReadyToProcess reports whether [hourBegin, hourBegin+1h) is complete and
// late-arriving logs have had time to land in S3.
func isHourReadyToProcess(hourBegin int64, now time.Time) bool {
	return now.Unix() >= hourBegin+hourSeconds+int64(hourProcessingDelay.Seconds())
}

func RunCluster(ctx context.Context, store storage.Storage, info types.ClusterInfo) {
	l := log.L().With(zap.String("cluster", info.StorageID))
	ts := time.Now().Unix() - config.Maxsecs
	ts -= ts % hourSeconds
	checkpointTs, err := loadCheckpoint(ctx, store, info.StorageID)
	if err != nil {
		l.Error("failed to load checkpoint", zap.Error(err))
		return
	}
	l.Info("run cluster", zap.Int64("checkpoint", checkpointTs), zap.Int64("next_hour", ts))
	if checkpointTs != 0 {
		ts = alignNextHour(checkpointTs)
	}
	for {
		select {
		case <-ctx.Done():
			return
		default:
		}
		now := time.Now()
		hourBegin := ts - ts%hourSeconds
		if !isHourReadyToProcess(hourBegin, now) {
			time.Sleep(time.Minute)
			continue
		}
		metrics.CurrentAnalyzeTasks.WithLabelValues(info.StorageID).Inc()
		ok := runClusterHourlyTask(ctx, store, info, hourBegin)
		metrics.CurrentAnalyzeTasks.WithLabelValues(info.StorageID).Dec()
		if ok {
			nextTs := hourBegin + hourSeconds
			if err := saveCheckpoint(ctx, store, info.StorageID, nextTs); err != nil {
				l.Error("failed to save checkpoint", zap.Error(err), zap.Int64("checkpoint", nextTs))
			}
			metrics.CheckpointTs.WithLabelValues(info.StorageID).Set(float64(nextTs))
			time.Sleep(3 * time.Second)
		}
		ts = hourBegin + hourSeconds
	}
}

func alignNextHour(checkpointTs int64) int64 {
	if checkpointTs%hourSeconds == 0 {
		return checkpointTs
	}
	return (checkpointTs/hourSeconds + 1) * hourSeconds
}

func runClusterHourlyTask(ctx context.Context, store storage.Storage, info types.ClusterInfo, hourBegin int64) bool {
	if config.ClusterTaskSem != nil {
		if err := config.ClusterTaskSem.Acquire(ctx, 1); err != nil {
			return false
		}
		defer config.ClusterTaskSem.Release(1)
	}

	l := log.L().With(zap.String("cluster", info.StorageID))
	utcHour := time.Unix(hourBegin, 0).UTC().Format("2006010215")
	hourEnd := hourBegin + hourSeconds - 1
	l.Info("run hourly task", zap.String("hour", utcHour), zap.Int64("hour_begin", hourBegin))

	bucket := config.BucketName()
	// Tiered conprof retention runs every hour, independent of log-stats idempotency.
	runConprofRetentionCleanup(ctx, store, bucket, info, hourBegin)

	statsKey := fmt.Sprintf("%s/data/%s/log-stats/1h/%s.db", config.S3Prefix(), info.StorageID, utcHour)
	size, err := store.ObjectSize(ctx, bucket, statsKey)
	if err != nil {
		l.Error("failed to check log-stats object", zap.Error(err))
		return false
	}
	if size > (10 * 1024) {
		l.Info("log-stats exists, skipped", zap.String("key", statsKey))
		return true
	}

	components, err := scanComponents(ctx, store, info)
	if err != nil {
		l.Error("failed to scan components", zap.Error(err))
		return false
	}
	if len(components) == 0 {
		// Still list profiles/regions for this hour even when there are no log components.
		recordHourlyAuxRawStats(ctx, store, bucket, info, hourBegin, hourEnd)
		l.Info("no components found for hour", zap.String("hour", utcHour))
		return true
	}

	dbFilepath := filepath.Join(config.LocalMergedFileDir, info.StorageID, "log-stats", "1h", utcHour+".db")
	if err := os.MkdirAll(path.Dir(dbFilepath), os.ModePerm); err != nil {
		l.Error("failed to make dirs", zap.Error(err))
		return false
	}
	db, err := InitDB(ctx, dbFilepath)
	if err != nil {
		l.Error("failed to initialize db", zap.Error(err))
		return false
	}

	parquetMgr, err := parquetpkg.NewManager(info, hourBegin)
	if err != nil {
		l.Error("failed to create parquet manager", zap.Error(err))
		_ = db.Close()
		return false
	}

	for _, component := range components {
		keys := findMatchedKeys(ctx, store, bucket, info, component, hourBegin, hourEnd)
		if len(keys) == 0 {
			continue
		}
		sort.Strings(keys)
		var filesOK, filesFailed, parseErrTotal int
		var parseErrSample string
		for _, key := range keys {
			parseErrs, sample, err := processRawLogObject(ctx, store, bucket, key, component, info, db, parquetMgr)
			parseErrTotal += parseErrs
			if parseErrSample == "" && sample != "" {
				parseErrSample = sample
			}
			if err != nil {
				filesFailed++
				l.Error("failed to process raw log object", zap.String("component", component), zap.String("key", key), zap.Error(err))
			} else {
				filesOK++
			}
			if uploadErr := parquetMgr.UploadClosedParts(ctx, store); uploadErr != nil {
				l.Error("failed to upload rolled parquet parts", zap.String("component", component), zap.Error(uploadErr))
				_ = db.Close()
				return false
			}
		}
		if parseErrTotal > 0 {
			l.Error("log parse errors in hourly component",
				zap.String("hour", utcHour),
				zap.String("component", component),
				zap.Int("count", parseErrTotal),
				zap.String("sample", parseErrSample),
			)
		}
		l.Info("component processed",
			zap.String("component", component),
			zap.Int("files_ok", filesOK),
			zap.Int("files_failed", filesFailed),
		)
	}

	if err := parquetMgr.FlushAndUpload(ctx, store); err != nil {
		l.Error("failed to upload parquet", zap.Error(err))
		_ = db.Close()
		return false
	}

	if err := db.Close(); err != nil {
		l.Warn("failed to close db", zap.Error(err))
	}
	f, err := os.Open(dbFilepath)
	if err != nil {
		l.Error("failed to open sqlite file", zap.Error(err))
		return false
	}
	defer f.Close()

	if err := store.SaveObject(ctx, bucket, statsKey, f); err != nil {
		l.Error("failed to upload sqlite db", zap.Error(err))
		return false
	}
	l.Info("log-stats uploaded", zap.String("key", statsKey))

	// profiles/ and regions/: metrics-only list (count + size), no ETL.
	// Done after log-stats upload so retries of a failed hour do not double-count.
	recordHourlyAuxRawStats(ctx, store, bucket, info, hourBegin, hourEnd)

	if err = deltalake.Sync2Deltalake(ctx, bucket, statsKey, info); err != nil {
		l.Error("failed to sync to deltalake", zap.Error(err))
	} else {
		l.Info("sync to deltalake", zap.String("key", statsKey))
	}
	l.Info("hourly task completed", zap.String("hour", utcHour))
	if err := os.Remove(dbFilepath); err != nil {
		l.Error("failed to remove sqlite file", zap.Error(err))
	}
	return true
}

func processRawLogObject(
	ctx context.Context,
	store storage.Storage,
	bucket, key, component string,
	info types.ClusterInfo,
	db *sql.DB,
	parquetMgr *parquetpkg.Manager,
) (parseErrCount int, parseErrSample string, err error) {
	if err := config.S3sem.Acquire(ctx, 1); err != nil {
		return 0, "", err
	}
	defer config.S3sem.Release(1)

	start := time.Now()
	obj, err := store.Object(ctx, bucket, key)
	if err != nil {
		metrics.RecordFileProcessed(info.StorageID, "error")
		return 0, "", err
	}
	defer obj.Reader.Close()

	reader, closer, err := openRawLogReader(key, obj.Reader)
	if err != nil {
		metrics.RecordFileProcessed(info.StorageID, "error")
		return 0, "", err
	}
	defer closer()

	fallbackTsUnix := timestampFromLogObjectKey(key)
	logReader := logutil.NewLogReader(reader, component, fallbackTsUnix)
	stats := types.NewStats()
	for {
		item, err := logReader.ReadNext()
		if err != nil {
			if err == logutil.ErrMalformedLine {
				continue
			}
			if err != io.EOF {
				metrics.RecordFileProcessed(info.StorageID, "error")
				return logReader.ParseErrorCount(), logReader.ParseErrorSample(), err
			}
			break
		}
		if item == nil {
			continue
		}
		stats.Merge(item, key)
		if item.TelemetryStr != "" {
			if err := insertTelemetryData(ctx, db, item, info.StorageID); err != nil {
				log.Error("failed to insert telemetry data", zap.Error(err))
			}
		}
		if err := parquetMgr.Write(component, parquetpkg.LogToRow(item, info.ClusterID, info.TenantID)); err != nil {
			metrics.RecordFileProcessed(info.StorageID, "error")
			return logReader.ParseErrorCount(), logReader.ParseErrorSample(), err
		}
	}
	if len(stats.Data) != 0 {
		if err := saveStats(ctx, db, stats, component, info); err != nil {
			return logReader.ParseErrorCount(), logReader.ParseErrorSample(), err
		}
	}
	metrics.RecordFileProcessed(info.StorageID, "success")
	metrics.RecordRawLogProcessed(info, component, rawLogObjectBytes(obj))
	metrics.FileProcessingDuration.WithLabelValues(info.StorageID, component).Observe(time.Since(start).Seconds())
	return logReader.ParseErrorCount(), logReader.ParseErrorSample(), nil
}

func timestampFromLogObjectKey(key string) int64 {
	base := path.Base(key)
	// Expected format: "<unixSeconds>-<uuid>.log" (or ".log.gz")
	ts, ok := unixTsFromFilename(base)
	if !ok {
		return 0
	}
	return ts
}

// recordHourlyAuxRawStats lists profiles/ and regions/ for the hour and records
// raw file count + compressed bytes. No download or further processing.
func recordHourlyAuxRawStats(
	ctx context.Context,
	store storage.Storage,
	bucket string,
	info types.ClusterInfo,
	hourBegin, hourEnd int64,
) {
	l := log.L().With(zap.String("cluster", info.StorageID))

	profileMetas := listUnixTsPrefixedObjects(ctx, store, bucket, strings.TrimSuffix(getProfilesPrefix(info), "/"), hourBegin, hourEnd)
	var profileBytes int64
	for _, meta := range profileMetas {
		metrics.RecordRawLogProcessed(info, "profiles", meta.Size)
		profileBytes += meta.Size
	}
	l.Info("profiles listed",
		zap.Int("files", len(profileMetas)),
		zap.Int64("bytes", profileBytes),
	)

	regionMetas := listRegionObjectsInHour(ctx, store, bucket, strings.TrimSuffix(getRegionsPrefix(info), "/"), hourBegin, hourEnd)
	var regionBytes int64
	for _, meta := range regionMetas {
		metrics.RecordRawLogProcessed(info, "regions", meta.Size)
		regionBytes += meta.Size
	}
	l.Info("regions listed",
		zap.Int("files", len(regionMetas)),
		zap.Int64("bytes", regionBytes),
	)
}

func rawLogObjectBytes(obj *storage.Object) int64 {
	if obj.ContentLength != nil {
		return *obj.ContentLength
	}
	return 0
}

func openRawLogReader(_ string, r io.Reader) (io.Reader, func(), error) {
	br := bufio.NewReader(r)
	header, err := br.Peek(2)
	if err != nil && err != io.EOF {
		return nil, nil, err
	}
	if len(header) >= 2 && header[0] == 0x1f && header[1] == 0x8b {
		gr, err := gzip.NewReader(br)
		if err != nil {
			return nil, nil, err
		}
		return gr, func() { _ = gr.Close() }, nil
	}
	return br, func() {}, nil
}