// Copyright 2023 PingCAP, Inc.
// Copyright 2015 CoreOS, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package file

import (
	"context"
	"encoding/binary"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"sync"
	"time"

	"github.com/pingcap/ticdc/pkg/common"
	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
	"github.com/pingcap/ticdc/pkg/errors"
	"github.com/pingcap/ticdc/pkg/fsutil"
	"github.com/pingcap/ticdc/pkg/metrics"
	"github.com/pingcap/ticdc/pkg/redo"
	"github.com/pingcap/ticdc/pkg/redo/codec"
	"github.com/pingcap/ticdc/pkg/redo/writer"
	"github.com/pingcap/ticdc/pkg/util"
	"github.com/pingcap/ticdc/pkg/uuid"
	"github.com/pingcap/tidb/br/pkg/storage"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/uber-go/atomic"
	pioutil "go.etcd.io/etcd/pkg/v3/ioutil"
	"golang.org/x/sync/errgroup"
)

//go:generate mockery --name=fileWriter --filename=file_mock.go --inpackage --quiet
type fileWriter interface {
	Run(ctx context.Context) error
	IsRunning() bool
	SyncWrite(event writer.RedoEvent) error
	GetInputCh() chan writer.RedoEvent
	Flush() error
	Close() error
	SetTableSchemaStore(*commonEvent.TableSchemaStore)
}

// fileWriter is a redo log event fileWriter which writes redo log events to a file.
type Writer struct {
	cfg     *writer.LogWriterConfig
	logType string
	op      *writer.LogWriterOptions
	inputCh chan writer.RedoEvent
	// maxCommitTS is the max commitTS among the events in one log file
	maxCommitTS atomic.Uint64
	// the ts used in file name
	commitTS atomic.Uint64
	// the ts send with the event
	eventCommitTS atomic.Uint64
	running       atomic.Bool
	size          int64
	file          *os.File
	// record the filepath that is being written, and has not been flushed
	ongoingFilePath string
	bw              *pioutil.PageWriter
	uint64buf       []byte
	storage         storage.ExternalStorage
	sync.RWMutex
	uuidGenerator uuid.Generator
	allocator     *fsutil.FileAllocator

	metricFsyncDuration    prometheus.Observer
	metricFlushAllDuration prometheus.Observer
	metricWriteBytes       prometheus.Gauge
	tableSchemaStore       *commonEvent.TableSchemaStore
}

// NewFileWriter return a file rotated writer, TODO: extract to a common rotate Writer
func NewFileWriter(
	ctx context.Context, cfg *writer.LogWriterConfig, logType string, opts ...writer.Option,
) (*Writer, error) {
	if cfg == nil {
		err := errors.New("FileWriterConfig can not be nil")
		return nil, errors.WrapError(errors.ErrRedoConfigInvalid, err)
	}

	var extStorage storage.ExternalStorage
	if cfg.UseExternalStorage {
		var err error
		extStorage, err = redo.InitExternalStorage(ctx, *cfg.URI)
		if err != nil {
			return nil, err
		}
	}

	op := &writer.LogWriterOptions{}
	for _, opt := range opts {
		opt(op)
	}

	w := &Writer{
		cfg:       cfg,
		logType:   logType,
		op:        op,
		inputCh:   make(chan writer.RedoEvent, redo.DefaultEncodingInputChanSize*util.GetOrZero(cfg.FlushWorkerNum)),
		uint64buf: make([]byte, 8),
		storage:   extStorage,

		metricFsyncDuration: metrics.RedoFsyncDurationHistogram.
			WithLabelValues(cfg.ChangeFeedID.Keyspace(), cfg.ChangeFeedID.Name(), logType),
		metricFlushAllDuration: metrics.RedoFlushAllDurationHistogram.
			WithLabelValues(cfg.ChangeFeedID.Keyspace(), cfg.ChangeFeedID.Name(), logType),
		metricWriteBytes: metrics.RedoWriteBytesGauge.
			WithLabelValues(cfg.ChangeFeedID.Keyspace(), cfg.ChangeFeedID.Name(), logType),
	}
	if w.op.GetUUIDGenerator != nil {
		w.uuidGenerator = w.op.GetUUIDGenerator()
	} else {
		w.uuidGenerator = uuid.NewGenerator()
	}

	if len(cfg.Dir) == 0 {
		return nil, errors.WrapError(errors.ErrRedoFileOp, errors.New("invalid redo dir path"))
	}

	err := os.MkdirAll(cfg.Dir, redo.DefaultDirMode)
	if err != nil {
		return nil, errors.WrapError(errors.ErrRedoFileOp,
			errors.Annotatef(err, "can't make dir: %s for redo writing", cfg.Dir))
	}

	// if we use S3 as the remote storage, a file allocator can be leveraged to
	// pre-allocate files for us.
	// TODO: test whether this improvement can also be applied to NFS.
	if w.cfg.UseExternalStorage {
		w.allocator = fsutil.NewFileAllocator(cfg.Dir, logType, cfg.MaxLogSizeInBytes)
	}

	w.running.Store(true)
	return w, nil
}

func (w *Writer) SetTableSchemaStore(tableSchemaStore *commonEvent.TableSchemaStore) {
	w.tableSchemaStore = tableSchemaStore
}

func (w *Writer) Run(ctx context.Context) error {
	eg, ctx := errgroup.WithContext(ctx)
	eg.Go(func() error {
		return w.encode(ctx)
	})
	return eg.Wait()
}

// Write implement write interface
// TODO: more general api with fileName generated by caller
func (w *Writer) Write(rawData []byte) (int, error) {
	w.Lock()
	defer w.Unlock()

	writeLen := int64(len(rawData))
	if writeLen > w.cfg.MaxLogSizeInBytes {
		return 0, errors.ErrRedoFileSizeExceed.GenWithStackByArgs(writeLen, w.cfg.MaxLogSizeInBytes)
	}

	if w.file == nil {
		if err := w.openNew(); err != nil {
			return 0, err
		}
	}

	if w.size+writeLen > w.cfg.MaxLogSizeInBytes {
		if err := w.rotate(); err != nil {
			return 0, err
		}
	}

	if w.maxCommitTS.Load() < w.eventCommitTS.Load() {
		w.maxCommitTS.Store(w.eventCommitTS.Load())
	}
	// ref: https://github.com/etcd-io/etcd/pull/5250
	lenField, padBytes := writer.EncodeFrameSize(len(rawData))
	if err := w.writeUint64(lenField, w.uint64buf); err != nil {
		return 0, err
	}

	if padBytes != 0 {
		rawData = append(rawData, make([]byte, padBytes)...)
	}

	n, err := w.bw.Write(rawData)
	if err != nil {
		return 0, err
	}
	w.metricWriteBytes.Add(float64(n))
	w.size += int64(n)

	return n, err
}

// AdvanceTs implement Advance interface
func (w *Writer) AdvanceTs(commitTs uint64) {
	w.eventCommitTS.Store(commitTs)
}

func (w *Writer) writeUint64(n uint64, buf []byte) error {
	binary.LittleEndian.PutUint64(buf, n)
	v, err := w.bw.Write(buf)
	w.metricWriteBytes.Add(float64(v))

	return err
}

// Close implements fileWriter.Close.
func (w *Writer) Close() error {
	w.Lock()
	defer w.Unlock()
	// always set to false when closed, since if having err may not get fixed just by retry
	defer w.running.Store(false)

	if w.allocator != nil {
		w.allocator.Close()
		w.allocator = nil
	}

	if !w.IsRunning() {
		return nil
	}

	metrics.RedoFlushAllDurationHistogram.
		DeleteLabelValues(w.cfg.ChangeFeedID.Keyspace(), w.cfg.ChangeFeedID.Name(), w.logType)
	metrics.RedoFsyncDurationHistogram.
		DeleteLabelValues(w.cfg.ChangeFeedID.Keyspace(), w.cfg.ChangeFeedID.Name(), w.logType)
	metrics.RedoWriteBytesGauge.
		DeleteLabelValues(w.cfg.ChangeFeedID.Keyspace(), w.cfg.ChangeFeedID.Name(), w.logType)

	ctx, cancel := context.WithTimeout(context.Background(), redo.CloseTimeout)
	defer cancel()
	return w.close(ctx)
}

// IsRunning implement IsRunning interface
func (w *Writer) IsRunning() bool {
	return w.running.Load()
}

func (w *Writer) GetInputCh() chan writer.RedoEvent {
	return w.inputCh
}

func (w *Writer) write(event writer.RedoEvent) error {
	rl := event.ToRedoLog()
	if rl.Type == commonEvent.RedoLogTypeDDL {
		rl.RedoDDL.SetTableSchemaStore(w.tableSchemaStore)
	}
	data, err := codec.MarshalRedoLog(rl, nil)
	if err != nil {
		return errors.WrapError(errors.ErrMarshalFailed, err)
	}
	w.AdvanceTs(rl.GetCommitTs())
	_, err = w.Write(data)
	if err != nil {
		return err
	}
	return nil
}

func (w *Writer) SyncWrite(event writer.RedoEvent) error {
	err := w.write(event)
	if err != nil {
		return err
	}
	err = w.Flush()
	if err != nil {
		return errors.Trace(err)
	}
	event.PostFlush()
	return nil
}

func (w *Writer) encode(ctx context.Context) error {
	d := time.Duration(util.GetOrZero(w.cfg.FlushIntervalInMs)) * time.Millisecond
	ticker := time.NewTicker(d)
	defer ticker.Stop()
	num := 0
	cacheEventPostFlush := make([]func(), 0, redo.DefaultFlushBatchSize)
	flush := func() error {
		err := w.Flush()
		if err != nil {
			return err
		}
		for _, fn := range cacheEventPostFlush {
			fn()
		}
		num = 0
		cacheEventPostFlush = cacheEventPostFlush[:0]
		return nil
	}
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-ticker.C:
		err := flush()
		if err != nil {
			return errors.Trace(err)
		}
	case e := <-w.inputCh:
		err := w.write(e)
		if err != nil {
			return err
		}
		num++
		if num > redo.DefaultFlushBatchSize {
			err := flush()
			if err != nil {
				return errors.Trace(err)
			}
			e.PostFlush()
		} else {
			cacheEventPostFlush = append(cacheEventPostFlush, e.PostFlush)
		}
	}
	return nil
}

func (w *Writer) close(ctx context.Context) error {
	if w.file == nil {
		return nil
	}

	if err := w.flush(); err != nil {
		return err
	}

	if w.cfg.UseExternalStorage {
		off, err := w.file.Seek(0, io.SeekCurrent)
		if err != nil {
			return err
		}
		// offset equals to 0 means that no written happened for current file,
		// we can simply return
		if off == 0 {
			return nil
		}
		// a file created by a file allocator needs to be truncated
		// to save disk space and network bandwidth.
		if err := w.file.Truncate(off); err != nil {
			return err
		}
	}

	// rename the file name from commitTs.log.tmp to maxCommitTS.log if closed safely
	// after rename, the file name could be used for search, since the ts is the max ts for all events in the file.
	w.commitTS.Store(w.maxCommitTS.Load())
	err := os.Rename(w.file.Name(), w.filePath())
	if err != nil {
		return errors.WrapError(errors.ErrRedoFileOp, err)
	}

	dirFile, err := os.Open(w.cfg.Dir)
	if err != nil {
		return errors.WrapError(errors.ErrRedoFileOp, err)
	}
	defer dirFile.Close()
	// sync the dir to guarantee the renamed file is persisted to disk.
	err = dirFile.Sync()
	if err != nil {
		return errors.WrapError(errors.ErrRedoFileOp, err)
	}

	// We only write content to S3 before closing the local file.
	// By this way, we no longer need renaming object in S3.
	if w.cfg.UseExternalStorage {
		err = w.writeToS3(ctx, w.ongoingFilePath)
		if err != nil {
			w.file.Close()
			w.file = nil
			return errors.WrapError(errors.ErrExternalStorageAPI, err)
		}
	}

	err = w.file.Close()
	w.file = nil
	return errors.WrapError(errors.ErrRedoFileOp, err)
}

func (w *Writer) getLogFileName() string {
	if w.op != nil && w.op.GetLogFileName != nil {
		return w.op.GetLogFileName()
	}
	uid := w.uuidGenerator.NewString()
	if common.DefaultKeyspaceName == w.cfg.ChangeFeedID.Keyspace() {
		return fmt.Sprintf(redo.RedoLogFileFormatV1,
			w.cfg.CaptureID, w.cfg.ChangeFeedID.Name(), w.logType,
			w.commitTS.Load(), uid, redo.LogEXT)
	}
	return fmt.Sprintf(redo.RedoLogFileFormatV2,
		w.cfg.CaptureID, w.cfg.ChangeFeedID.Keyspace(), w.cfg.ChangeFeedID.Name(),
		w.logType, w.commitTS.Load(), uid, redo.LogEXT)
}

// filePath always creates a new, unique file path, note this function is not
// thread-safe, writer needs to ensure lock is acquired when calling it.
func (w *Writer) filePath() string {
	fp := filepath.Join(w.cfg.Dir, w.getLogFileName())
	w.ongoingFilePath = fp
	return fp
}

func openTruncFile(name string) (*os.File, error) {
	return os.OpenFile(name, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, redo.DefaultFileMode)
}

func (w *Writer) openNew() error {
	err := os.MkdirAll(w.cfg.Dir, redo.DefaultDirMode)
	if err != nil {
		return errors.WrapError(errors.ErrRedoFileOp,
			errors.Annotatef(err, "can't make dir: %s for new redo logfile", w.cfg.Dir))
	}

	// reset ts used in file name when new file
	var f *os.File
	if w.allocator == nil {
		w.commitTS.Store(w.eventCommitTS.Load())
		w.maxCommitTS.Store(w.eventCommitTS.Load())
		path := w.filePath() + redo.TmpEXT
		f, err = openTruncFile(path)
		if err != nil {
			return errors.WrapError(errors.ErrRedoFileOp,
				errors.Annotate(err, "can't open new redolog file"))
		}
	} else {
		// if there is a file allocator, we use the pre-created file
		// supplied by the allocator to boost performance
		f, err = w.allocator.Open()
		if err != nil {
			return errors.WrapError(errors.ErrRedoFileOp,
				errors.Annotate(err, "can't open new redolog file with file allocator"))
		}
	}
	w.file = f
	w.size = 0
	err = w.newPageWriter()
	if err != nil {
		return err
	}
	return nil
}

func (w *Writer) newPageWriter() error {
	offset, err := w.file.Seek(0, io.SeekCurrent)
	if err != nil {
		return errors.WrapError(errors.ErrRedoFileOp, err)
	}
	w.bw = pioutil.NewPageWriter(w.file, redo.PageBytes, int(offset))

	return nil
}

func (w *Writer) rotate() error {
	ctx, cancel := context.WithTimeout(context.Background(), redo.DefaultTimeout)
	defer cancel()
	if err := w.close(ctx); err != nil {
		return err
	}
	return w.openNew()
}

// flushAndRotateFile flushes the file to disk and rotate it if S3 storage is used.
func (w *Writer) flushAndRotateFile() error {
	if w.file == nil {
		return nil
	}

	start := time.Now()
	err := w.flush()
	if err != nil {
		return err
	}

	if !w.cfg.UseExternalStorage {
		return nil
	}

	if w.size == 0 {
		return nil
	}

	// for s3 storage, when the file is flushed to disk, we need an immediate
	// file rotate. Otherwise, the existing file content would be repeatedly written to S3,
	// which could cause considerable network bandwidth waste.
	err = w.rotate()
	if err != nil {
		return nil
	}
	w.metricFlushAllDuration.Observe(time.Since(start).Seconds())

	return err
}

// Flush implement Flush interface
func (w *Writer) Flush() error {
	w.Lock()
	defer w.Unlock()

	return w.flushAndRotateFile()
}

func (w *Writer) flush() error {
	if w.file == nil {
		return nil
	}

	n, err := w.bw.FlushN()
	w.metricWriteBytes.Add(float64(n))
	if err != nil {
		return errors.WrapError(errors.ErrRedoFileOp, err)
	}

	start := time.Now()
	err = w.file.Sync()
	w.metricFsyncDuration.Observe(time.Since(start).Seconds())

	return errors.WrapError(errors.ErrRedoFileOp, err)
}

func (w *Writer) writeToS3(ctx context.Context, name string) error {
	fileData, err := os.ReadFile(name)
	if err != nil {
		return errors.WrapError(errors.ErrRedoFileOp, err)
	}

	// Key in s3: aws.String(rs.options.Prefix + name), prefix should be changefeed name
	err = w.storage.WriteFile(ctx, filepath.Base(name), fileData)
	if err != nil {
		return errors.WrapError(errors.ErrExternalStorageAPI, err)
	}

	// in case the page cache piling up triggered the OS memory reclaming which may cause
	// I/O latency spike, we mandatorily drop the page cache of the file when it is successfully
	// written to S3.
	err = fsutil.DropPageCache(name)
	if err != nil {
		return errors.WrapError(errors.ErrRedoFileOp, err)
	}

	return nil
}