package gcs

import (
	"context"
	"errors"
	"fmt"
	"io"
	"os"
	"path"

	"cloud.google.com/go/storage"
	"github.com/sirupsen/logrus"
	"google.golang.org/api/iterator"
	"google.golang.org/api/option"

	"github.com/tidbcloud/observability/backup-worker/pkg/common"
	"github.com/tidbcloud/observability/backup-worker/pkg/configs"
)

type Client struct {
	cli *storage.Client
	cfg configs.GcsSink
	rp  configs.RetryPolicy
}

// New creates a Gcs client with flexible credential configuration.
// Credential priority:
// 1. CredentialsJSON: Inline JSON credentials (from environment variable)
// 2. CredentialsFile: File path to credentials JSON (from Kubernetes secret)
// 3. Application Default Credentials: Automatic on GCP (GKE, Cloud Run, etc.)
//
// ADC is recommended for production as it eliminates credential management
// and enables automatic credential rotation via workload identity.
func New(ctx context.Context, cfg configs.GcsSink, rp configs.RetryPolicy) (*Client, error) {
	var opts []option.ClientOption

	// Support both inline JSON and file-based credentials.
	// Inline JSON is useful for containerized deployments where credentials
	// are injected via environment variables.
	if cfg.CredentialsJSON != "" {
		opts = append(opts, option.WithCredentialsJSON([]byte(cfg.CredentialsJSON)))
	} else if cfg.CredentialsFile != "" {
		opts = append(opts, option.WithCredentialsFile(cfg.CredentialsFile))
	}

	cli, err := storage.NewClient(ctx, opts...)
	if err != nil {
		return nil, fmt.Errorf("create gcs client: %w", err)
	}

	logrus.Infof("Gcs client created: bucket=%s", cfg.Bucket)
	return &Client{cli: cli, cfg: cfg, rp: rp}, nil
}

func (c *Client) Type() string {
	return common.StorageGCS
}

// CheckObject checks whether an object already exists in the bucket.
//
// Used by MetricWorker to skip shards whose target object was already uploaded
// in a previous run, making backups resumable after interruption.
func (c *Client) CheckObject(ctx context.Context, key string) (bool, error) {
	normalizeKey := c.normalizePath(key)
	_, err := c.cli.Bucket(c.cfg.Bucket).Object(normalizeKey).Attrs(ctx)
	if err == nil {
		return true, nil
	}
	if errors.Is(err, storage.ErrObjectNotExist) {
		return false, nil
	}
	return false, fmt.Errorf("check gcs object: %w", err)
}

// SinkObject uploads a local file to GCS with automatic retries.
//
// The GCS SDK handles resumable uploads automatically, which:
// - Survives network interruptions
// - Supports files larger than memory
// - Provides progress tracking for monitoring
func (c *Client) SinkObject(ctx context.Context, key, localPath string) error {
	normalizeKey := c.normalizePath(key)
	return c.rp.Do(ctx, func() error {
		return c.sinkObject(ctx, normalizeKey, localPath)
	})
}

func (c *Client) sinkObject(ctx context.Context, key, localPath string) error {
	f, err := os.Open(localPath)
	if err != nil {
		return fmt.Errorf("open local file: %w", err)
	}
	defer f.Close()

	w := c.cli.Bucket(c.cfg.Bucket).Object(key).NewWriter(ctx)
	if _, err := io.Copy(w, f); err != nil {
		return fmt.Errorf("write to gcs: %w", err)
	}
	// Close commits the upload; an error here means the data did not persist.
	if err := w.Close(); err != nil {
		return fmt.Errorf("close gcs writer: %w", err)
	}
	return nil
}

// ListObjects iterates through all objects under the given prefix.
func (c *Client) ListObjects(ctx context.Context, prefix string) ([]common.ObjectInfo, error) {
	var objects []common.ObjectInfo
	it := c.cli.Bucket(c.cfg.Bucket).Objects(ctx, &storage.Query{Prefix: prefix})
	for {
		attrs, err := it.Next()
		if errors.Is(err, iterator.Done) {
			break
		}
		if err != nil {
			return nil, fmt.Errorf("list gcs objects '%s': %w", prefix, err)
		}
		objects = append(objects, common.ObjectInfo{
			Name: attrs.Name,
			Size: attrs.Size,
		})
	}
	return objects, nil
}

// CopyObject copies an object from another bucket into this client's bucket.
// GCS CopierFrom handles large files internally via the rewrite API,
// which automatically chunks and retries — no explicit multipart management needed.
func (c *Client) CopyObject(ctx context.Context, srcBucket, srcKey, destKey string) error {
	return c.rp.Do(ctx, func() error {
		src := c.cli.Bucket(srcBucket).Object(srcKey)
		dst := c.cli.Bucket(c.cfg.Bucket).Object(destKey)
		if _, err := dst.CopierFrom(src).Run(ctx); err != nil {
			return fmt.Errorf("copy gcs object %s/%s -> %s/%s: %w", srcBucket, srcKey, c.cfg.Bucket, destKey, err)
		}
		return nil
	})
}

func (c *Client) Close() error {
	return c.cli.Close()
}

func (c *Client) normalizePath(key string) string {
	return path.Join(c.cfg.PathPrefix, key)
}