package app

import (
	"compress/gzip"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"math/big"
	"os"
	"path"
	"strings"
	"time"

	"hyapp/pkg/appcode"
	"hyapp/pkg/logx"
	"hyapp/services/wallet-service/internal/config"
	mysqlstorage "hyapp/services/wallet-service/internal/storage/mysql"
)

const (
	walletOutboxArchiveSchema        = "hyapp.wallet_outbox.archive"
	walletOutboxArchiveSchemaVersion = 1
	walletOutboxArchiveManifestState = "DATA_VERIFIED"
	walletOutboxArchiveVerifiedState = "VERIFIED"
)

type walletOutboxArchiveManifest struct {
	Schema              string                                 `json:"schema"`
	SchemaVersion       int                                    `json:"schema_version"`
	State               string                                 `json:"state"`
	AppCode             string                                 `json:"app_code"`
	BatchID             string                                 `json:"batch_id"`
	RowCount            int                                    `json:"row_count"`
	FirstCursor         mysqlstorage.WalletOutboxArchiveCursor `json:"first_cursor"`
	LastCursor          mysqlstorage.WalletOutboxArchiveCursor `json:"last_cursor"`
	MinCreatedAtMS      int64                                  `json:"min_created_at_ms"`
	MaxCreatedAtMS      int64                                  `json:"max_created_at_ms"`
	MinUpdatedAtMS      int64                                  `json:"min_updated_at_ms"`
	MaxUpdatedAtMS      int64                                  `json:"max_updated_at_ms"`
	AvailableDeltaTotal string                                 `json:"available_delta_total"`
	FrozenDeltaTotal    string                                 `json:"frozen_delta_total"`
	NDJSONSHA256        string                                 `json:"ndjson_sha256"`
	GzipSHA256          string                                 `json:"gzip_sha256"`
	DataCRC64           string                                 `json:"data_crc64"`
	UncompressedBytes   int64                                  `json:"uncompressed_bytes"`
	CompressedBytes     int64                                  `json:"compressed_bytes"`
	DataObjectKey       string                                 `json:"data_object_key"`
}

type walletOutboxArchiveSuccess struct {
	Schema         string `json:"schema"`
	SchemaVersion  int    `json:"schema_version"`
	State          string `json:"state"`
	BatchID        string `json:"batch_id"`
	ManifestSHA256 string `json:"manifest_sha256"`
	ManifestCRC64  string `json:"manifest_crc64"`
}

type walletOutboxArchiveArtifact struct {
	manifest          walletOutboxArchiveManifest
	manifestBody      []byte
	manifestSHA256    string
	successBody       []byte
	successSHA256     string
	dataFilePath      string
	dataObjectKey     string
	manifestObjectKey string
	successObjectKey  string
}

func (a *App) startWalletOutboxArchiveWorker(ctx context.Context) {
	a.workers.Add(1)
	go func() {
		defer a.workers.Done()
		ticker := time.NewTicker(a.outboxArchiveCfg.PollInterval)
		defer ticker.Stop()
		for {
			if err := a.runWalletOutboxArchiveRound(ctx); err != nil && !errors.Is(err, context.Canceled) {
				logx.Error(ctx, "wallet_outbox_archive_round_failed", err)
			}
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
			}
		}
	}()
}

func (a *App) runWalletOutboxArchiveRound(ctx context.Context) error {
	if a.mysqlRepo == nil {
		return nil
	}
	queryCtx, cancel := context.WithTimeout(ctx, a.outboxArchiveCfg.QueryTimeout)
	appCodes, err := a.mysqlRepo.ListWalletOutboxAppCodes(queryCtx)
	cancel()
	if err != nil {
		return err
	}
	cutoffMS := time.Now().UTC().Add(-a.outboxArchiveCfg.MaxAge).UnixMilli()
	for _, appCode := range appCodes {
		var appErr error
		if a.outboxArchiveCfg.Mode == config.OutboxArchiveModeDryRun {
			appErr = a.inspectWalletOutboxArchivePage(ctx, appCode, cutoffMS)
		} else {
			appErr = a.archiveWalletOutboxAppRound(ctx, appCode, cutoffMS)
		}
		if appErr != nil {
			// 单租户归档失败不得阻塞其他 App；失败 App 不写 receipt，下一轮仍从同一游标重试。
			logx.Error(ctx, "wallet_outbox_archive_app_failed", appErr, slog.String("app_code", appCode))
		}
	}
	return nil
}

func (a *App) inspectWalletOutboxArchivePage(ctx context.Context, targetAppCode string, cutoffMS int64) error {
	tenantCtx := appcode.WithContext(ctx, targetAppCode)
	cursor := a.outboxArchiveDryRunCursors[targetAppCode]
	queryCtx, cancel := context.WithTimeout(tenantCtx, a.outboxArchiveCfg.QueryTimeout)
	records, err := a.mysqlRepo.ListDeliveredWalletOutboxArchiveBatch(queryCtx, cursor, cutoffMS, a.outboxArchiveCfg.QueryBatchSize)
	cancel()
	if err != nil || len(records) == 0 {
		return err
	}
	first := records[0]
	last := records[len(records)-1]
	a.outboxArchiveDryRunCursors[targetAppCode] = mysqlstorage.WalletOutboxArchiveCursor{UpdatedAtMS: last.UpdatedAtMS, EventID: last.EventID}
	logx.Info(ctx, "wallet_outbox_archive_planned",
		slog.String("app_code", targetAppCode),
		slog.Int("row_count", len(records)),
		slog.Int64("cutoff_ms", cutoffMS),
		slog.Int64("first_updated_at_ms", first.UpdatedAtMS),
		slog.Int64("last_updated_at_ms", last.UpdatedAtMS),
		slog.String("state", "PLANNED"),
		slog.Bool("purge_enabled", false),
	)
	return nil
}

func (a *App) archiveWalletOutboxAppRound(ctx context.Context, targetAppCode string, cutoffMS int64) error {
	tenantCtx := appcode.WithContext(ctx, targetAppCode)
	lockCtx, lockCancel := context.WithTimeout(tenantCtx, a.outboxArchiveCfg.QueryTimeout)
	release, acquired, err := a.mysqlRepo.AcquireWalletOutboxArchiveAppLock(lockCtx)
	lockCancel()
	if err != nil || !acquired {
		return err
	}
	defer release()
	queryCtx, cancel := context.WithTimeout(tenantCtx, a.outboxArchiveCfg.QueryTimeout)
	cursor, err := a.mysqlRepo.LatestVerifiedWalletOutboxArchiveCursor(queryCtx)
	cancel()
	if err != nil {
		return err
	}
	archivedRows := 0
	for archivedRows < a.outboxArchiveCfg.RoundMaxRows {
		segmentLimit := min(a.outboxArchiveCfg.SegmentMaxRows, a.outboxArchiveCfg.RoundMaxRows-archivedRows)
		records, err := a.collectWalletOutboxArchiveSegment(tenantCtx, cursor, cutoffMS, segmentLimit)
		if err != nil || len(records) == 0 {
			return err
		}
		if err := a.archiveWalletOutboxRecords(tenantCtx, targetAppCode, records); err != nil {
			return err
		}
		last := records[len(records)-1]
		cursor = mysqlstorage.WalletOutboxArchiveCursor{UpdatedAtMS: last.UpdatedAtMS, EventID: last.EventID}
		archivedRows += len(records)
		if len(records) < segmentLimit {
			break
		}
	}
	return nil
}

func (a *App) collectWalletOutboxArchiveSegment(ctx context.Context, cursor mysqlstorage.WalletOutboxArchiveCursor, cutoffMS int64, segmentLimit int) ([]mysqlstorage.WalletOutboxArchiveRecord, error) {
	records := make([]mysqlstorage.WalletOutboxArchiveRecord, 0, segmentLimit)
	partitionDay := ""
	var uncompressedBytes int64
	for len(records) < segmentLimit {
		pageLimit := min(a.outboxArchiveCfg.QueryBatchSize, segmentLimit-len(records))
		queryCtx, cancel := context.WithTimeout(ctx, a.outboxArchiveCfg.QueryTimeout)
		page, err := a.mysqlRepo.ListDeliveredWalletOutboxArchiveBatch(queryCtx, cursor, cutoffMS, pageLimit)
		cancel()
		if err != nil {
			return nil, err
		}
		if len(page) == 0 {
			break
		}
		for _, record := range page {
			recordDay := time.UnixMilli(record.UpdatedAtMS).UTC().Format("2006-01-02")
			if partitionDay == "" {
				partitionDay = recordDay
			}
			// 一个对象严格属于一个 UTC delivered 日，保证按日盘点/恢复不会遗漏跨日批次里的前半段。
			if recordDay != partitionDay {
				return records, nil
			}
			line, err := json.Marshal(record)
			if err != nil {
				return nil, err
			}
			recordBytes := int64(len(line) + 1)
			// 至少允许单行进入批次；超大单行仍由 operation timeout 和临时盘容量形成外层保护。
			if len(records) > 0 && uncompressedBytes+recordBytes > a.outboxArchiveCfg.SegmentMaxUncompressedBytes {
				return records, nil
			}
			records = append(records, record)
			uncompressedBytes += recordBytes
			if len(records) == segmentLimit {
				return records, nil
			}
		}
		last := records[len(records)-1]
		cursor = mysqlstorage.WalletOutboxArchiveCursor{UpdatedAtMS: last.UpdatedAtMS, EventID: last.EventID}
		if len(page) < pageLimit {
			break
		}
		if a.outboxArchiveCfg.BatchPause > 0 {
			select {
			case <-ctx.Done():
				return nil, ctx.Err()
			case <-time.After(a.outboxArchiveCfg.BatchPause):
			}
		}
	}
	return records, nil
}

func (a *App) archiveWalletOutboxRecords(tenantCtx context.Context, targetAppCode string, records []mysqlstorage.WalletOutboxArchiveRecord) error {
	if a.outboxArchiveStore == nil {
		return fmt.Errorf("wallet outbox archive store is not configured")
	}
	artifact, err := buildWalletOutboxArchiveArtifact(a.outboxArchiveCfg, records)
	if err != nil {
		return err
	}
	defer artifact.remove()
	operationCtx, operationCancel := context.WithTimeout(tenantCtx, a.outboxArchiveCfg.OperationTimeout)
	defer operationCancel()
	// 只有 data 读回通过后才上传 manifest，manifest 读回通过后才写 _SUCCESS；
	// 任意中断都不会产生 MySQL verified receipt，后续可用相同内容键幂等重试。
	dataCRC64, err := a.outboxArchiveStore.putGzipAndVerify(operationCtx, artifact.dataObjectKey, artifact.dataFilePath, artifact.manifest.CompressedBytes, artifact.manifest.GzipSHA256, artifact.manifest.NDJSONSHA256)
	if err != nil {
		return err
	}
	if err := artifact.finalizeManifest(dataCRC64); err != nil {
		return err
	}
	manifestCRC64, err := a.outboxArchiveStore.putBytesAndVerify(operationCtx, artifact.manifestObjectKey, artifact.manifestBody, "application/json", artifact.manifestSHA256)
	if err != nil {
		return err
	}
	if err := artifact.finalizeSuccess(manifestCRC64); err != nil {
		return err
	}
	successCRC64, err := a.outboxArchiveStore.putBytesAndVerify(operationCtx, artifact.successObjectKey, artifact.successBody, "application/json", artifact.successSHA256)
	if err != nil {
		return err
	}
	nowMS := time.Now().UTC().UnixMilli()
	members := make([]mysqlstorage.WalletOutboxArchiveMember, 0, len(records))
	for _, record := range records {
		// 精确成员证明与 VERIFIED receipt 同事务落库；purge 不再用游标范围猜测某一行是否真的存在于 COS 对象。
		members = append(members, mysqlstorage.WalletOutboxArchiveMember{
			AppCode: record.AppCode, EventID: record.EventID, UpdatedAtMS: record.UpdatedAtMS,
		})
	}
	receiptCtx, receiptCancel := context.WithTimeout(tenantCtx, a.outboxArchiveCfg.QueryTimeout)
	err = a.mysqlRepo.RecordVerifiedWalletOutboxArchive(receiptCtx, mysqlstorage.WalletOutboxArchiveReceipt{
		AppCode: targetAppCode, BatchID: artifact.manifest.BatchID, State: walletOutboxArchiveVerifiedState,
		DataObjectKey: artifact.dataObjectKey, ManifestObjectKey: artifact.manifestObjectKey, SuccessObjectKey: artifact.successObjectKey,
		SchemaVersion: artifact.manifest.SchemaVersion, RowCount: artifact.manifest.RowCount,
		FirstCursor: artifact.manifest.FirstCursor, LastCursor: artifact.manifest.LastCursor,
		MinCreatedAtMS: artifact.manifest.MinCreatedAtMS, MaxCreatedAtMS: artifact.manifest.MaxCreatedAtMS,
		AvailableDeltaTotal: artifact.manifest.AvailableDeltaTotal, FrozenDeltaTotal: artifact.manifest.FrozenDeltaTotal,
		NDJSONSHA256: artifact.manifest.NDJSONSHA256, GzipSHA256: artifact.manifest.GzipSHA256, ManifestSHA256: artifact.manifestSHA256,
		DataCRC64: dataCRC64, ManifestCRC64: manifestCRC64, SuccessCRC64: successCRC64,
		UncompressedBytes: artifact.manifest.UncompressedBytes, CompressedBytes: artifact.manifest.CompressedBytes, VerifiedAtMS: nowMS,
		Members: members,
	})
	receiptCancel()
	if err != nil {
		return err
	}
	logx.Info(tenantCtx, "wallet_outbox_archive_verified",
		slog.String("app_code", targetAppCode), slog.String("batch_id", artifact.manifest.BatchID),
		slog.Int("row_count", artifact.manifest.RowCount), slog.String("state", walletOutboxArchiveVerifiedState),
		slog.Bool("purge_enabled", false),
	)
	return nil
}

func buildWalletOutboxArchiveArtifact(cfg config.OutboxArchiveConfig, records []mysqlstorage.WalletOutboxArchiveRecord) (walletOutboxArchiveArtifact, error) {
	if len(records) == 0 {
		return walletOutboxArchiveArtifact{}, fmt.Errorf("wallet outbox archive batch is empty")
	}
	tempFile, err := os.CreateTemp(cfg.TempDir, "wallet-outbox-*.ndjson.gz")
	if err != nil {
		return walletOutboxArchiveArtifact{}, err
	}
	tempPath := tempFile.Name()
	removeOnError := true
	defer func() {
		if removeOnError {
			_ = tempFile.Close()
			_ = os.Remove(tempPath)
		}
	}()
	// CreateTemp 默认 0600，仍显式收紧权限，避免自定义 umask 导致账务快照被同机其他用户读取。
	if err := tempFile.Chmod(0o600); err != nil {
		return walletOutboxArchiveArtifact{}, err
	}
	gzipHash := sha256.New()
	gzipWriter := gzip.NewWriter(io.MultiWriter(tempFile, gzipHash))
	gzipWriter.Header.ModTime = time.Unix(0, 0).UTC()
	gzipWriter.Header.OS = 255
	ndjsonHash := sha256.New()
	countedNDJSON := &countingWriter{writer: io.MultiWriter(gzipWriter, ndjsonHash)}
	availableTotal := new(big.Int)
	frozenTotal := new(big.Int)
	minCreatedAtMS, maxCreatedAtMS := records[0].CreatedAtMS, records[0].CreatedAtMS
	partitionDay := time.UnixMilli(records[0].UpdatedAtMS).UTC().Format("2006-01-02")
	for _, record := range records {
		if record.AppCode != records[0].AppCode || record.Status != "delivered" {
			return walletOutboxArchiveArtifact{}, fmt.Errorf("wallet outbox archive batch crosses app boundary or contains non-delivered row")
		}
		if time.UnixMilli(record.UpdatedAtMS).UTC().Format("2006-01-02") != partitionDay {
			return walletOutboxArchiveArtifact{}, fmt.Errorf("wallet outbox archive batch crosses UTC delivered day")
		}
		line, err := json.Marshal(record)
		if err != nil {
			return walletOutboxArchiveArtifact{}, err
		}
		line = append(line, '\n')
		if _, err := countedNDJSON.Write(line); err != nil {
			return walletOutboxArchiveArtifact{}, err
		}
		availableTotal.Add(availableTotal, big.NewInt(record.AvailableDelta))
		frozenTotal.Add(frozenTotal, big.NewInt(record.FrozenDelta))
		if record.CreatedAtMS < minCreatedAtMS {
			minCreatedAtMS = record.CreatedAtMS
		}
		if record.CreatedAtMS > maxCreatedAtMS {
			maxCreatedAtMS = record.CreatedAtMS
		}
	}
	if err := gzipWriter.Close(); err != nil {
		return walletOutboxArchiveArtifact{}, err
	}
	if err := tempFile.Sync(); err != nil {
		return walletOutboxArchiveArtifact{}, err
	}
	info, err := tempFile.Stat()
	if err != nil {
		return walletOutboxArchiveArtifact{}, err
	}
	if err := tempFile.Close(); err != nil {
		return walletOutboxArchiveArtifact{}, err
	}
	first, last := records[0], records[len(records)-1]
	firstCursor := mysqlstorage.WalletOutboxArchiveCursor{UpdatedAtMS: first.UpdatedAtMS, EventID: first.EventID}
	lastCursor := mysqlstorage.WalletOutboxArchiveCursor{UpdatedAtMS: last.UpdatedAtMS, EventID: last.EventID}
	ndjsonSHA256 := hex.EncodeToString(ndjsonHash.Sum(nil))
	gzipSHA256 := hex.EncodeToString(gzipHash.Sum(nil))
	batchHash := sha256.Sum256([]byte(strings.Join([]string{first.AppCode, fmt.Sprint(len(records)), fmt.Sprint(firstCursor.UpdatedAtMS), firstCursor.EventID, fmt.Sprint(lastCursor.UpdatedAtMS), lastCursor.EventID, ndjsonSHA256}, "\n")))
	batchID := hex.EncodeToString(batchHash[:])
	objectDirectory := path.Join(cfg.ObjectPrefix, "app_code="+first.AppCode, "delivered_day="+partitionDay, "batch_id="+batchID)
	dataObjectKey := path.Join(objectDirectory, "data.ndjson.gz")
	manifest := walletOutboxArchiveManifest{
		Schema: walletOutboxArchiveSchema, SchemaVersion: walletOutboxArchiveSchemaVersion, State: walletOutboxArchiveManifestState,
		AppCode: first.AppCode, BatchID: batchID, RowCount: len(records), FirstCursor: firstCursor, LastCursor: lastCursor,
		MinCreatedAtMS: minCreatedAtMS, MaxCreatedAtMS: maxCreatedAtMS, MinUpdatedAtMS: first.UpdatedAtMS, MaxUpdatedAtMS: last.UpdatedAtMS,
		AvailableDeltaTotal: availableTotal.String(), FrozenDeltaTotal: frozenTotal.String(), NDJSONSHA256: ndjsonSHA256, GzipSHA256: gzipSHA256,
		UncompressedBytes: countedNDJSON.count, CompressedBytes: info.Size(), DataObjectKey: dataObjectKey,
	}
	removeOnError = false
	return walletOutboxArchiveArtifact{
		manifest: manifest, dataFilePath: tempPath,
		dataObjectKey: dataObjectKey, manifestObjectKey: path.Join(objectDirectory, "manifest.json"), successObjectKey: path.Join(objectDirectory, "_SUCCESS"),
	}, nil
}

func (a *walletOutboxArchiveArtifact) finalizeManifest(dataCRC64 string) error {
	dataCRC64 = strings.TrimSpace(dataCRC64)
	if dataCRC64 == "" {
		return fmt.Errorf("wallet outbox archive data CRC64 is empty")
	}
	a.manifest.DataCRC64 = dataCRC64
	manifestBody, err := json.Marshal(a.manifest)
	if err != nil {
		return err
	}
	a.manifestBody = append(manifestBody, '\n')
	manifestHash := sha256.Sum256(a.manifestBody)
	a.manifestSHA256 = hex.EncodeToString(manifestHash[:])
	return nil
}

func (a *walletOutboxArchiveArtifact) finalizeSuccess(manifestCRC64 string) error {
	manifestCRC64 = strings.TrimSpace(manifestCRC64)
	if manifestCRC64 == "" {
		return fmt.Errorf("wallet outbox archive manifest CRC64 is empty")
	}
	successBody, err := json.Marshal(walletOutboxArchiveSuccess{
		Schema: walletOutboxArchiveSchema, SchemaVersion: walletOutboxArchiveSchemaVersion,
		State: walletOutboxArchiveVerifiedState, BatchID: a.manifest.BatchID,
		ManifestSHA256: a.manifestSHA256, ManifestCRC64: manifestCRC64,
	})
	if err != nil {
		return err
	}
	a.successBody = append(successBody, '\n')
	successHash := sha256.Sum256(a.successBody)
	a.successSHA256 = hex.EncodeToString(successHash[:])
	return nil
}

func (a walletOutboxArchiveArtifact) remove() {
	if strings.TrimSpace(a.dataFilePath) != "" {
		_ = os.Remove(a.dataFilePath)
	}
}

type countingWriter struct {
	writer io.Writer
	count  int64
}

func (w *countingWriter) Write(body []byte) (int, error) {
	written, err := w.writer.Write(body)
	w.count += int64(written)
	return written, err
}