package backend

import (
	"bytes"
	"context"
	"database/sql"
	"errors"
	"fmt"
	"io"
	"time"

	"github.com/pingcap/failpoint"
	"go.uber.org/zap"

	"github.com/mem9-ai/drive9/pkg/datastore"
	"github.com/mem9-ai/drive9/pkg/logger"
	"github.com/mem9-ai/drive9/pkg/meta"
	"github.com/mem9-ai/drive9/pkg/pathutil"
	"github.com/mem9-ai/drive9/pkg/s3client"
)

const maxAppendLogObjectBytes int64 = 5_000_000_000

type AppendLogResult struct {
	Revision int64
	Size     int64
}

type appendLogExactBodyReader struct {
	body     io.Reader
	expected int64
	read     int64
}

func (r *appendLogExactBodyReader) Read(p []byte) (int, error) {
	if r.expected < 0 {
		return 0, s3client.ErrAppendBodySizeMismatch
	}
	remaining := r.expected + 1 - r.read
	if remaining <= 0 {
		return 0, s3client.ErrAppendBodySizeMismatch
	}
	if int64(len(p)) > remaining {
		p = p[:remaining]
	}
	n, err := r.body.Read(p)
	r.read += int64(n)
	if r.read > r.expected {
		return n, s3client.ErrAppendBodySizeMismatch
	}
	if err == io.EOF && r.read != r.expected {
		return n, s3client.ErrAppendBodySizeMismatch
	}
	return n, err
}

func (r *appendLogExactBodyReader) VerifyEOF() error {
	if r.read != r.expected {
		return s3client.ErrAppendBodySizeMismatch
	}
	var probe [1]byte
	for {
		n, err := r.body.Read(probe[:])
		if n > 0 {
			return s3client.ErrAppendBodySizeMismatch
		}
		if err == io.EOF {
			return nil
		}
		if err != nil {
			return err
		}
	}
}

// AppendLog appends one immutable tail to an append-log object. It only reports
// success after the physical offset append and the logical metadata CAS commit.
func (b *Dat9Backend) AppendLog(ctx context.Context, path string, body io.Reader, size, expectedRevision, expectedSize int64) (*AppendLogResult, error) {
	if !b.AppendLogEnabled() {
		return nil, ErrAppendLogUnsupported
	}
	start := time.Now()
	stage := start
	statMs, precheckMs, s3Ms, logMs := int64(-1), int64(-1), int64(-1), int64(-1)
	txMs := int64(-1)
	defer func() {
		logger.InfoBenchTiming(ctx, "server_append_log_timing",
			zap.String("path", path),
			zap.Int64("size", size),
			zap.Int64("stat_ms", statMs),
			zap.Int64("precheck_ms", precheckMs),
			zap.Int64("s3_ms", s3Ms),
			zap.Int64("tx_ms", txMs),
			zap.Int64("log_ms", logMs),
			zap.Int64("total_ms", time.Since(start).Milliseconds()))
	}()
	if size < 0 || expectedRevision < 0 || expectedSize < 0 || expectedSize > maxAppendLogObjectBytes-size {
		return nil, ErrAppendLogTooLarge
	}
	path, err := pathutil.Canonicalize(path)
	if err != nil {
		return nil, err
	}
	if err := rejectRootFileNodePath(path); err != nil {
		return nil, err
	}

	nf, err := b.store.StatAppendLogTarget(ctx, path)
	if errors.Is(err, datastore.ErrNotFound) {
		if expectedRevision != 0 || expectedSize != 0 {
			return nil, datastore.ErrRevisionConflict
		}
		return b.createAppendLog(ctx, path, body, size)
	}
	if err != nil {
		return nil, err
	}
	if nf.Node.IsDirectory || nf.File == nil || nf.File.ContentLayout != datastore.ContentLayoutAppendLog || nf.File.StorageType != datastore.StorageS3 {
		return nil, ErrAppendLogUnsupported
	}
	if nf.File.Revision != expectedRevision || nf.File.SizeBytes != expectedSize {
		return nil, datastore.ErrRevisionConflict
	}
	if size == 0 {
		return nil, fmt.Errorf("%w: zero-byte append is only valid for create", ErrAppendLogInvalidRequest)
	}
	statMs = time.Since(stage).Milliseconds()
	stage = time.Now()
	if err := b.ensureFileSizeQuota(ctx, expectedSize+size); err != nil {
		return nil, err
	}
	// Do predictable, read-only quota admission before mutating the live
	// object. The in-transaction check below remains the authoritative
	// confirmation. The legacy tenant-DB quota path keeps its tx-based
	// precheck.
	if b.UseServerQuota() {
		if err := b.ensureStorageQuotaPrecheck(ctx, size); err != nil {
			return nil, err
		}
	} else {
		if err := b.store.InTx(ctx, func(tx *sql.Tx) error {
			return b.ensureStorageQuota(ctx, tx, path, expectedSize+size)
		}); err != nil {
			return nil, err
		}
	}
	precheckMs = time.Since(stage).Milliseconds()
	stage = time.Now()
	if err := b.appendLogS3.AppendObject(ctx, nf.File.StorageRef, body, size, expectedSize); err != nil {
		if errors.Is(err, s3client.ErrAppendOffsetMismatch) || errors.Is(err, s3client.ErrAppendTooManyParts) {
			return nil, b.rebaseAppendLog(ctx, path, nf.File, errors.Is(err, s3client.ErrAppendTooManyParts))
		}
		return nil, err
	}
	s3Ms = time.Since(stage).Milliseconds()
	stage = time.Now()

	var revision int64
	if err := b.store.InTx(ctx, func(tx *sql.Tx) error {
		live, err := b.ensureAppendLogPathBindingTx(ctx, tx, path, nf.File.FileID)
		if err != nil {
			return err
		}
		currentSize := int64(0)
		if live.File != nil {
			currentSize = live.File.SizeBytes
		}
		if err := b.ensureStorageQuotaFromSnapshotTx(ctx, tx, path, currentSize, expectedSize+size); err != nil {
			return err
		}
		var injectedFinalTxErr error
		failpoint.InjectCall("appendLogFinalTxFail", path, &injectedFinalTxErr)
		if injectedFinalTxErr != nil {
			return injectedFinalTxErr
		}
		var txErr error
		// The final transaction may fail (quota/conflict) after the S3 append
		// succeeded, leaving orphan tail bytes beyond the metadata size. Reads
		// truncate by metadata size; the next append at the same offset gets
		// InvalidWriteOffset and the rebase path rebuilds the ref from the
		// metadata-size prefix.
		revision, txErr = b.store.AppendLogContentIfMatchTx(tx, live, expectedRevision, expectedSize, nf.File.StorageRef, size)
		return txErr
	}); err != nil {
		return nil, err
	}
	txMs = time.Since(stage).Milliseconds()
	stage = time.Now()
	quotaCtx, cancel := postCommitQuotaMutationContext()
	defer cancel()
	if err := b.recordCentralFileOverwriteMutation(quotaCtx, nf.File.FileID, expectedSize, nf.File.ContentType, expectedSize+size, nf.File.ContentType, StorageTypeS3); err != nil {
		return nil, postCommitQuotaMutationError("record append-log quota mutation", err)
	}
	logMs = time.Since(stage).Milliseconds()
	return &AppendLogResult{Revision: revision, Size: expectedSize + size}, nil
}

func (b *Dat9Backend) rebaseAppendLog(ctx context.Context, path string, f *datastore.File, copyWhole bool) error {
	if f == nil || !b.AppendLogEnabled() {
		return ErrAppendLogUnsupported
	}
	if f.SizeBytes < 0 || f.SizeBytes > maxAppendLogObjectBytes {
		return ErrAppendLogTooLarge
	}
	nextRef := b.nextAppendLogRef(f.FileID)
	publishCtx, cancel := context.WithTimeout(ctx, FailedWriteObjectGCPublishTimeout)
	defer cancel()
	if err := b.prepareFailedWriteObjectGCWithLayout(publishCtx, nextRef, f.FileID, datastore.ContentLayoutAppendLog); err != nil {
		return err
	}
	var err error
	if copyWhole {
		err = b.appendLogS3.CopyAppendLogObject(publishCtx, f.StorageRef, nextRef)
	} else if f.SizeBytes == 0 {
		err = b.appendLogS3.PutAppendLogObject(publishCtx, nextRef, nil, 0)
	} else {
		var prefix io.ReadCloser
		prefix, err = b.appendLogS3.GetObjectRange(publishCtx, f.StorageRef, 0, f.SizeBytes-1)
		if err == nil {
			defer func() { _ = prefix.Close() }()
			err = b.appendLogS3.PutAppendLogObject(publishCtx, nextRef, prefix, f.SizeBytes)
		}
	}
	if err != nil {
		b.failFailedWriteObjectGC(ctx, nextRef, err)
		return err
	}
	if err := b.beginFailedWriteObjectGCPublish(publishCtx, nextRef); err != nil {
		b.cleanupAppendLogFailedWriteObjectGC(ctx, nextRef)
		return err
	}
	if _, err := b.enqueueObjectGCCandidateWithLayoutCtx(publishCtx, f.StorageRef, datastore.ContentLayoutAppendLog, meta.ObjectGCReasonOverwrite, f.FileID); err != nil {
		b.cleanupAppendLogFailedWriteObjectGC(ctx, nextRef)
		return fmt.Errorf("prepare append-log old-ref cleanup: %w", err)
	}
	var nextRevision int64
	err = b.store.InTx(publishCtx, func(tx *sql.Tx) error {
		if _, err := b.ensureAppendLogPathBindingTx(publishCtx, tx, path, f.FileID); err != nil {
			return err
		}
		var txErr error
		nextRevision, txErr = b.store.ReplaceAppendLogRefIfMatchTx(tx, f.FileID, f.Revision, f.SizeBytes, f.StorageRef, nextRef)
		return txErr
	})
	if err != nil {
		b.reconcileAppendLogFailedWriteCandidate(ctx, nextRef, err)
		return err
	}
	b.completeFailedWriteObjectGC(ctx, nextRef)
	_ = nextRevision
	return ErrAppendLogRebased
}

func (b *Dat9Backend) ensureAppendLogPathBindingTx(ctx context.Context, tx *sql.Tx, path, expectedFileID string) (*datastore.NodeWithFile, error) {
	live, err := b.store.StatForUpdateTx(ctx, tx, path)
	if errors.Is(err, datastore.ErrNotFound) {
		return nil, fmt.Errorf("%w: append-log path no longer exists", datastore.ErrRevisionConflict)
	}
	if err != nil {
		return nil, err
	}
	if live.Node.IsDirectory || datastore.NodeFileID(live) != expectedFileID {
		return nil, fmt.Errorf("%w: append-log path no longer names the original file", datastore.ErrRevisionConflict)
	}
	return live, nil
}

// reconcileAppendLogFailedWriteCandidate leaves an unknown commit outcome
// reachable by object GC rather than permanently retaining its publish lease.
func (b *Dat9Backend) reconcileAppendLogFailedWriteCandidate(ctx context.Context, storageRef string, metadataErr error) {
	if errors.Is(metadataErr, datastore.ErrCommitOutcomeUnknown) {
		b.failFailedWriteObjectGC(ctx, storageRef, metadataErr)
		return
	}
	b.cleanupAppendLogFailedWriteObjectGC(ctx, storageRef)
}

func (b *Dat9Backend) createAppendLog(ctx context.Context, path string, body io.Reader, size int64) (*AppendLogResult, error) {
	if err := b.ensureFileSizeQuota(ctx, size); err != nil {
		return nil, err
	}
	if err := b.store.InTx(ctx, func(tx *sql.Tx) error {
		if err := b.ensureCreateStorageQuota(ctx, tx, size); err != nil {
			return err
		}
		return b.ensureFileCountQuotaServer(ctx, tx, 1)
	}); err != nil {
		return nil, err
	}
	fileID := b.genID()
	ref := b.nextAppendLogRef(fileID)
	publishCtx, cancel := context.WithTimeout(ctx, FailedWriteObjectGCPublishTimeout)
	defer cancel()
	if err := b.prepareFailedWriteObjectGCWithLayout(publishCtx, ref, fileID, datastore.ContentLayoutAppendLog); err != nil {
		return nil, fmt.Errorf("prepare append-log failed-write cleanup: %w", err)
	}
	if err := b.appendLogS3.PutAppendLogObject(publishCtx, ref, body, size); err != nil {
		b.failFailedWriteObjectGC(ctx, ref, err)
		return nil, err
	}
	if err := b.beginFailedWriteObjectGCPublish(publishCtx, ref); err != nil {
		b.cleanupAppendLogFailedWriteObjectGC(ctx, ref)
		return nil, err
	}
	now := time.Now().UTC()
	if err := b.store.InTx(publishCtx, func(tx *sql.Tx) error {
		if err := b.ensureCreateStorageQuota(ctx, tx, size); err != nil {
			return err
		}
		if err := b.ensureFileCountQuotaServer(ctx, tx, 1); err != nil {
			return err
		}
		if err := b.store.InsertFileTx(tx, &datastore.File{
			FileID:                fileID,
			StorageType:           datastore.StorageS3,
			ContentLayout:         datastore.ContentLayoutAppendLog,
			StorageRef:            ref,
			StorageEncryptionMode: datastore.StorageEncryptionNone,
			SizeBytes:             size,
			Revision:              1,
			Status:                datastore.StatusConfirmed,
			CreatedAt:             now,
			ConfirmedAt:           &now,
		}); err != nil {
			return err
		}
		if err := b.store.EnsureParentDirsTx(tx, path, b.genID); err != nil {
			return err
		}
		return b.store.InsertNodeTx(tx, &datastore.FileNode{NodeID: b.genID(), Path: path, ParentPath: pathutil.ParentPath(path), Name: pathutil.BaseName(path), FileID: fileID, CreatedAt: now})
	}); err != nil {
		b.reconcileAppendLogFailedWriteCandidate(ctx, ref, err)
		return nil, err
	}
	b.completeFailedWriteObjectGC(ctx, ref)
	quotaCtx, cancelQuota := postCommitQuotaMutationContext()
	defer cancelQuota()
	if err := b.recordCentralFileCreateMutation(quotaCtx, fileID, size, "", StorageTypeS3); err != nil {
		return nil, postCommitQuotaMutationError("record append-log create quota mutation", err)
	}
	return &AppendLogResult{Revision: 1, Size: size}, nil
}

// RewriteAppendLog replaces an existing append-log generation with a streamed
// full image. It is the safe fallback for truncates and non-tail writes.
func (b *Dat9Backend) RewriteAppendLog(ctx context.Context, path string, body io.Reader, size, expectedRevision int64) (*AppendLogResult, error) {
	if !b.AppendLogEnabled() {
		return nil, ErrAppendLogUnsupported
	}
	if size < 0 || size > maxAppendLogObjectBytes || expectedRevision < 0 {
		return nil, ErrAppendLogTooLarge
	}
	path, err := pathutil.Canonicalize(path)
	if err != nil {
		return nil, err
	}
	nf, err := b.store.Stat(ctx, path)
	if err != nil {
		return nil, err
	}
	if nf.Node.IsDirectory || nf.File == nil || nf.File.ContentLayout != datastore.ContentLayoutAppendLog || nf.File.StorageType != datastore.StorageS3 {
		return nil, ErrAppendLogUnsupported
	}
	if nf.File.Revision != expectedRevision {
		return nil, datastore.ErrRevisionConflict
	}
	if err := b.ensureFileSizeQuota(ctx, size); err != nil {
		return nil, err
	}
	if err := b.store.InTx(ctx, func(tx *sql.Tx) error {
		return b.ensureStorageQuota(ctx, tx, path, size)
	}); err != nil {
		return nil, err
	}
	nextRef := b.nextAppendLogRef(nf.File.FileID)
	publishCtx, cancel := context.WithTimeout(ctx, FailedWriteObjectGCPublishTimeout)
	defer cancel()
	if err := b.prepareFailedWriteObjectGCWithLayout(publishCtx, nextRef, nf.File.FileID, datastore.ContentLayoutAppendLog); err != nil {
		return nil, err
	}
	if body == nil {
		body = bytes.NewReader(nil)
	}
	exactBody := &appendLogExactBodyReader{body: body, expected: size}
	if err := b.appendLogS3.PutAppendLogObject(publishCtx, nextRef, exactBody, size); err != nil {
		b.failFailedWriteObjectGC(ctx, nextRef, err)
		return nil, err
	}
	if err := exactBody.VerifyEOF(); err != nil {
		b.cleanupAppendLogFailedWriteObjectGC(ctx, nextRef)
		return nil, err
	}
	if err := b.beginFailedWriteObjectGCPublish(publishCtx, nextRef); err != nil {
		b.cleanupAppendLogFailedWriteObjectGC(ctx, nextRef)
		return nil, err
	}
	if _, err := b.enqueueObjectGCCandidateWithLayoutCtx(publishCtx, nf.File.StorageRef, datastore.ContentLayoutAppendLog, meta.ObjectGCReasonOverwrite, nf.File.FileID); err != nil {
		b.cleanupAppendLogFailedWriteObjectGC(ctx, nextRef)
		return nil, fmt.Errorf("prepare append-log old-ref cleanup: %w", err)
	}
	var revision int64
	if err := b.store.InTx(publishCtx, func(tx *sql.Tx) error {
		if _, err := b.ensureAppendLogPathBindingTx(publishCtx, tx, path, nf.File.FileID); err != nil {
			return err
		}
		if err := b.ensureStorageQuota(ctx, tx, path, size); err != nil {
			return err
		}
		var txErr error
		revision, txErr = b.store.ReplaceAppendLogContentIfMatchTx(tx, nf.File.FileID, expectedRevision, nf.File.SizeBytes, nf.File.StorageRef, nextRef, size)
		return txErr
	}); err != nil {
		b.reconcileAppendLogFailedWriteCandidate(ctx, nextRef, err)
		return nil, err
	}
	b.completeFailedWriteObjectGC(ctx, nextRef)
	quotaCtx, cancelQuota := postCommitQuotaMutationContext()
	defer cancelQuota()
	if err := b.recordCentralFileOverwriteMutation(quotaCtx, nf.File.FileID, nf.File.SizeBytes, nf.File.ContentType, size, nf.File.ContentType, StorageTypeS3); err != nil {
		return nil, postCommitQuotaMutationError("record append-log rewrite quota mutation", err)
	}
	return &AppendLogResult{Revision: revision, Size: size}, nil
}

func (b *Dat9Backend) nextAppendLogRef(fileID string) string {
	return "append-log/" + b.storageNamespaceID + "/" + fileID + "/" + b.genID()
}