package expoimport

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"time"
	"xprem/config"
	"xprem/internal/bucket"
	cache2 "xprem/internal/cache"
	"xprem/internal/crypto"
	"xprem/internal/dashboard"
	"xprem/internal/jobs"
	"xprem/internal/providers/expo"
	"xprem/internal/repository"
	"xprem/internal/types"
	update2 "xprem/internal/update"
	"xprem/internal/validation"

	"github.com/google/uuid"
	"github.com/riverqueue/river"
	"github.com/riverqueue/river/rivertype"
	"golang.org/x/sync/errgroup"
)

var ErrHistoryImportAlreadyRunning = errors.New("an update history import is already running for this app")

var ErrHistoryJobNotFound = errors.New("this history import job does not exist")

const MaxHistoryImportGroups = 50

const expoHistoryJobKind = "expo-history-import"

// The update groups are snapshotted at enqueue time so a retry never needs
// the Expo token, which is not stored.
type expoHistoryArgs struct {
	AppId  string                 `json:"appId" river:"unique"`
	Total  int                    `json:"total"`
	Groups [][]expo.HistoryUpdate `json:"groups"`
}

func (expoHistoryArgs) Kind() string { return expoHistoryJobKind }

func (expoHistoryArgs) InsertOpts() river.InsertOpts {
	return river.InsertOpts{
		MaxAttempts: 5,
		UniqueOpts: river.UniqueOpts{
			ByArgs: true,
			// The default state list also blocks a new job while a finished
			// one is retained; this list only blocks while one is live.
			ByState: []rivertype.JobState{
				rivertype.JobStateAvailable,
				rivertype.JobStatePending,
				rivertype.JobStateRunning,
				rivertype.JobStateRetryable,
				rivertype.JobStateScheduled,
			},
		},
	}
}

type expoHistoryWorker struct {
	river.WorkerDefaults[expoHistoryArgs]
	service *Service
}

func (w *expoHistoryWorker) Timeout(*river.Job[expoHistoryArgs]) time.Duration {
	return time.Hour
}

func (w *expoHistoryWorker) Work(ctx context.Context, job *river.Job[expoHistoryArgs]) error {
	return w.service.copyHistory(ctx, job.Args.AppId, jobs.NewTracker(job.ID), job.Args.Groups)
}

func RegisterWorker(workers *river.Workers, service *Service) {
	river.AddWorker(workers, &expoHistoryWorker{service: service})
}

type HistoryJobStatus struct {
	State           string   `json:"state"`
	Total           int      `json:"total"`
	Processed       int      `json:"processed"`
	Imported        int      `json:"imported"`
	Skipped         []string `json:"skipped,omitempty"`
	Error           string   `json:"error,omitempty"`
	CancelRequested bool     `json:"cancelRequested"`
}

func historyJobStatus(job *rivertype.JobRow) *HistoryJobStatus {
	var args expoHistoryArgs
	_ = json.Unmarshal(job.EncodedArgs, &args)
	output := jobs.OutputOf(job)
	return &HistoryJobStatus{
		State:           jobs.UIState(job),
		Total:           args.Total,
		Processed:       output.Processed,
		Imported:        output.Succeeded,
		Skipped:         output.Warnings,
		Error:           jobs.LastError(job),
		CancelRequested: jobs.CancelRequested(job),
	}
}

func (s *Service) StartHistoryImport(ctx context.Context, auth types.Auth, expoAppId string, limit int) (string, error) {
	if !config.IsDBMode() {
		return "", repository.ErrNotSupportedInStatelessMode
	}
	if err := requireExpoAuth(auth); err != nil {
		return "", err
	}
	parsedId, err := uuid.Parse(strings.TrimSpace(expoAppId))
	if err != nil {
		return "", validation.Errorf("expoAppId", "must be the Expo project UUID (app.json, extra.eas.projectId)")
	}
	if limit < 1 || limit > MaxHistoryImportGroups {
		return "", validation.Errorf("limit", "must be between 1 and %d", MaxHistoryImportGroups)
	}
	appId := parsedId.String()
	if _, err := s.apps.GetAppByID(ctx, appId); err != nil {
		return "", validation.Errorf("expoAppId", "this app is not imported yet, import it first")
	}

	groups, err := expo.FetchUpdateGroups(ctx, auth, appId, limit)
	if err != nil {
		return "", err
	}
	total := 0
	for _, group := range groups {
		total += len(group)
	}

	jobId, err := s.jobs.Enqueue(ctx, expoHistoryArgs{AppId: appId, Total: total, Groups: groups})
	if errors.Is(err, jobs.ErrAlreadyRunning) {
		return "", ErrHistoryImportAlreadyRunning
	}
	return jobId, err
}

func (s *Service) GetAppHistoryJob(ctx context.Context, appId string) (string, *HistoryJobStatus, bool) {
	parsed, err := uuid.Parse(strings.TrimSpace(appId))
	if err != nil {
		return "", nil, false
	}
	job, err := s.jobs.LatestByArg(ctx, expoHistoryJobKind, "appId", parsed.String())
	if err != nil || job == nil {
		return "", nil, false
	}
	return strconv.FormatInt(job.ID, 10), historyJobStatus(job), true
}

func (s *Service) GetHistoryJobStatus(ctx context.Context, jobId string) (*HistoryJobStatus, bool) {
	job, err := s.jobs.Get(ctx, jobId)
	if err != nil || job == nil {
		return nil, false
	}
	return historyJobStatus(job), true
}

// The cancel lands once the update being copied is finished; canceling an
// already finished job is a no-op.
func (s *Service) CancelHistoryJob(ctx context.Context, jobId string) error {
	found, err := s.jobs.Cancel(ctx, jobId)
	if err != nil {
		return err
	}
	if !found {
		return ErrHistoryJobNotFound
	}
	return nil
}

type branchRuntime struct {
	branch  string
	runtime string
}

// copyHistory copies the oldest fetched group first, so a stopped job leaves
// a consistent prefix of history.
func (s *Service) copyHistory(ctx context.Context, appId string, tracker *jobs.Tracker, groups [][]expo.HistoryUpdate) error {
	touched := make(map[branchRuntime]bool)
	defer s.invalidateHistoryServingCaches(appId, touched)

	for groupIndex := len(groups) - 1; groupIndex >= 0; groupIndex-- {
		for _, historyUpdate := range groups[groupIndex] {
			skipReason, err := s.importHistoryUpdate(ctx, appId, historyUpdate, touched)
			// A canceled context outranks whatever the update reported.
			if ctxErr := ctx.Err(); ctxErr != nil {
				return ctxErr
			}
			if err != nil {
				return fmt.Errorf("failed while importing update %s: %w", historyUpdate.Id, err)
			}
			if skipReason != "" {
				tracker.Skip(ctx, fmt.Sprintf("update %s (%s): %s", historyUpdate.Id, historyUpdate.Platform, skipReason))
			} else {
				tracker.Succeed(ctx)
			}
		}
	}
	return nil
}

// importHistoryUpdate returns a non-empty skip reason for problems scoped to
// this update, an error for failures that must stop the job.
func (s *Service) importHistoryUpdate(ctx context.Context, appId string, historyUpdate expo.HistoryUpdate, touched map[branchRuntime]bool) (string, error) {
	platform, ok := parseHistoryPlatform(historyUpdate.Platform)
	if !ok {
		return fmt.Sprintf("platform %q is not supported", historyUpdate.Platform), nil
	}
	// The branch name is validated by the upsert below; the runtime version has
	// no other checkpoint before the insert.
	if err := validation.Name("runtimeVersion", historyUpdate.RuntimeVersion); err != nil {
		return fmt.Sprintf("runtime version %q: %s", historyUpdate.RuntimeVersion, validationMessage(err)), nil
	}
	if historyUpdate.CodeSigned {
		return "signed with Expo code signing, its signature cannot be carried over", nil
	}
	createdAt, err := time.Parse(time.RFC3339, historyUpdate.CreatedAt)
	if err != nil {
		return fmt.Sprintf("unparseable publication date %q", historyUpdate.CreatedAt), nil
	}
	expoUpdateUUID, err := uuid.Parse(historyUpdate.Id)
	if err != nil {
		return fmt.Sprintf("unparseable update id %q", historyUpdate.Id), nil
	}

	// Same id scheme as GenerateUpdateTimestamp, but derived from the original
	// EAS publication instant so the branch timeline keeps its order.
	updateId := createdAt.UnixMilli()*10 + historyPlatformDigit(platform)
	// An occupied slot: already imported, or published locally at this exact
	// instant; writing this update's files would overwrite its own.
	occupied, err := s.updateRepo.UpdateExists(ctx, appId, historyUpdate.BranchName, updateId)
	if err != nil {
		return "", err
	}
	if occupied {
		return "an update already exists at this instant on this branch", nil
	}

	if err := s.branches.UpsertBranchAndRuntimeVersion(ctx, appId, historyUpdate.BranchName, historyUpdate.RuntimeVersion); err != nil {
		var valErr *validation.Error
		if errors.As(err, &valErr) {
			return fmt.Sprintf("branch %q: %s", historyUpdate.BranchName, valErr.Message), nil
		}
		return "", fmt.Errorf("failed to upsert branch %q and runtime version %q: %w", historyUpdate.BranchName, historyUpdate.RuntimeVersion, err)
	}

	params := repository.ImportUpdateParams{
		AppId:          appId,
		UpdateId:       updateId,
		BranchName:     historyUpdate.BranchName,
		RuntimeVersion: historyUpdate.RuntimeVersion,
		Platform:       platform,
		CommitHash:     historyUpdate.GitCommitHash,
		Message:        historyUpdate.Message,
		CreatedAt:      createdAt,
		CheckedAt:      createdAt,
	}
	if group, err := uuid.Parse(historyUpdate.Group); err == nil {
		groupStr := group.String()
		params.PublishGroup = &groupStr
	}

	if historyUpdate.IsRollBack {
		params.UpdateType = types.Rollback
		if _, err := s.updateRepo.ImportUpdate(ctx, params); err != nil {
			return "", err
		}
		touched[branchRuntime{branch: historyUpdate.BranchName, runtime: historyUpdate.RuntimeVersion}] = true
		return "", nil
	}

	if historyUpdate.ManifestPermalink == "" {
		return "the update carries no manifest permalink", nil
	}
	served, err := expo.FetchServedManifest(ctx, historyUpdate.ManifestPermalink)
	if err != nil {
		return err.Error(), nil
	}
	if served.Manifest.LaunchAsset.Url == "" {
		return "the manifest carries no launch asset", nil
	}

	update := types.Update{
		AppId:          appId,
		Branch:         historyUpdate.BranchName,
		RuntimeVersion: historyUpdate.RuntimeVersion,
		UpdateId:       strconv.FormatInt(updateId, 10),
	}
	// Blobs first: on a skip nothing has landed in the update folder yet, and
	// cas blobs are shared so they are never cleaned up.
	mapping, skipReason, err := s.storeHistoryBlobs(ctx, appId, served)
	if err != nil {
		return "", err
	}
	if skipReason != "" {
		return skipReason, nil
	}
	if err := s.writeHistoryConfigFiles(ctx, update, platform, historyUpdate, &served.Manifest); err != nil {
		s.deleteHistoryUpdateFolder(ctx, update)
		return "", err
	}

	expoUpdateUUIDStr := expoUpdateUUID.String()
	params.UpdateType = types.NormalUpdate
	params.UpdateUUID = &expoUpdateUUIDStr
	params.AssetMapping = mapping
	inserted, err := s.updateRepo.ImportUpdate(ctx, params)
	if err != nil {
		// Without the row the config files are unreachable orphans.
		s.deleteHistoryUpdateFolder(ctx, update)
		return "", err
	}
	touched[branchRuntime{branch: historyUpdate.BranchName, runtime: historyUpdate.RuntimeVersion}] = true
	if !inserted {
		log.Printf("[expo-import] timeline slot %s/%s already occupied; bucket files for update %s may have been overwritten", historyUpdate.BranchName, update.UpdateId, historyUpdate.Id)
		return "an update already exists at this instant on this branch; its files may have been overwritten", nil
	}
	return "", nil
}

const historyDownloadConcurrency = 8

// skipUpdate is a per-update problem: it skips the update instead of failing the job.
type skipUpdate struct{ reason string }

func (e *skipUpdate) Error() string { return e.reason }

// storeHistoryBlobs puts every asset of the served manifest in cas/ and
// returns the update's asset mapping. EAS hashes are base64url SHA-256, the
// exact cas key format, so a hash already in cas is not even downloaded.
// Download and integrity problems return a skip reason, cas writes an error.
func (s *Service) storeHistoryBlobs(ctx context.Context, appId string, served *expo.ServedManifest) (*types.UpdateAssetMapping, string, error) {
	manifest := &served.Manifest
	mapping := &types.UpdateAssetMapping{
		LaunchAsset: shapeHistoryAsset(manifest.LaunchAsset, true),
		Assets:      make([]types.ShapedAsset, len(manifest.Assets)),
	}
	sources := []expo.HistoryAsset{manifest.LaunchAsset}
	for i, asset := range manifest.Assets {
		mapping.Assets[i] = shapeHistoryAsset(asset, false)
		sources = append(sources, asset)
	}

	var mu sync.Mutex
	hashByURL := make(map[string]string, len(sources))
	group, groupCtx := errgroup.WithContext(ctx)
	group.SetLimit(historyDownloadConcurrency)
	seenURLs := make(map[string]bool, len(sources))
	for _, asset := range sources {
		if asset.Url == "" {
			return nil, fmt.Sprintf("asset %q carries no download URL", asset.Key), nil
		}
		if seenURLs[asset.Url] {
			continue
		}
		seenURLs[asset.Url] = true
		group.Go(func() error {
			hash, err := s.ensureHistoryBlob(groupCtx, appId, asset, served.AssetRequestHeaders[asset.Key])
			if err != nil {
				return err
			}
			mu.Lock()
			hashByURL[asset.Url] = hash
			mu.Unlock()
			return nil
		})
	}
	if err := group.Wait(); err != nil {
		var skip *skipUpdate
		if errors.As(err, &skip) {
			return nil, skip.reason, nil
		}
		return nil, "", err
	}

	// An asset the manifest did not hash gets the one computed from its bytes.
	mapping.LaunchAsset.Hash = hashByURL[manifest.LaunchAsset.Url]
	for i, asset := range manifest.Assets {
		mapping.Assets[i].Hash = hashByURL[asset.Url]
	}
	return mapping, "", nil
}

// ensureHistoryBlob returns the cas hash of the asset, downloading and storing
// it only when the blob is not already there.
func (s *Service) ensureHistoryBlob(ctx context.Context, appId string, asset expo.HistoryAsset, headers map[string]string) (string, error) {
	declaredHash := asset.Hash
	if bucket.ValidateBlobHash(declaredHash) != nil {
		declaredHash = ""
	}
	if declaredHash != "" {
		exists, err := s.blobStore.Exists(ctx, appId, declaredHash)
		if err != nil {
			return "", fmt.Errorf("failed to check cas for asset %q: %w", asset.Key, err)
		}
		if exists {
			return declaredHash, nil
		}
	}
	data, err := expo.DownloadAsset(ctx, asset.Url, headers)
	if err != nil {
		return "", &skipUpdate{reason: err.Error()}
	}
	hash, err := crypto.CreateHash(data, "sha256", "base64")
	if err != nil {
		return "", err
	}
	computedHash := crypto.GetBase64URLEncoding(hash)
	if declaredHash != "" && computedHash != declaredHash {
		return "", &skipUpdate{reason: fmt.Sprintf("asset %q does not match its manifest hash", asset.Key)}
	}
	if err := s.blobStore.Put(ctx, appId, computedHash, bytes.NewReader(data)); err != nil {
		return "", fmt.Errorf("failed to write asset %q into cas: %w", asset.Key, err)
	}
	return computedHash, nil
}

// shapeHistoryAsset mirrors the publish-side shaping: the launch asset serves
// as .bundle whatever EAS called it.
func shapeHistoryAsset(asset expo.HistoryAsset, isLaunchAsset bool) types.ShapedAsset {
	shaped := types.ShapedAsset{
		Hash:          asset.Hash,
		Key:           asset.Key,
		FileExtension: asset.FileExtension,
		ContentType:   asset.ContentType,
	}
	if isLaunchAsset {
		shaped.FileExtension = ".bundle"
		if shaped.ContentType == "" {
			shaped.ContentType = "application/javascript"
		}
	}
	return shaped
}

// writeHistoryConfigFiles writes the update folder's config files; the assets
// themselves live in cas/.
func (s *Service) writeHistoryConfigFiles(ctx context.Context, update types.Update, platform types.Platform, historyUpdate expo.HistoryUpdate, manifest *expo.HistoryManifest) error {
	platformMetadata := types.PlatformMetadata{
		Bundle: "bundles/" + string(platform) + "-" + historyAssetFileName(manifest.LaunchAsset, 0) + ".bundle",
	}
	seenPaths := map[string]bool{platformMetadata.Bundle: true}
	for index, asset := range manifest.Assets {
		path := "assets/" + historyAssetFileName(asset, index)
		if seenPaths[path] {
			continue
		}
		seenPaths[path] = true
		platformMetadata.Assets = append(platformMetadata.Assets, types.Asset{
			Path: path,
			Ext:  strings.TrimPrefix(asset.FileExtension, "."),
		})
	}

	metadata := types.MetadataObject{Version: 0, Bundler: "metro"}
	if platform == types.PlatformIOS {
		metadata.FileMetadata.IOS = platformMetadata
	} else {
		metadata.FileMetadata.Android = platformMetadata
	}
	metadataBytes, err := json.Marshal(metadata)
	if err != nil {
		return err
	}
	if err := s.updateStore.PutFile(ctx, update, "metadata.json", bytes.NewReader(metadataBytes)); err != nil {
		return fmt.Errorf("failed to write metadata.json into the bucket: %w", err)
	}

	if expoClient := manifest.ExpoClientConfig(); expoClient != nil {
		expoConfigBytes, err := json.Marshal(expoClient)
		if err != nil {
			return err
		}
		if err := s.updateStore.PutFile(ctx, update, "expoConfig.json", bytes.NewReader(expoConfigBytes)); err != nil {
			return fmt.Errorf("failed to write expoConfig.json into the bucket: %w", err)
		}
	}

	// The stored UUID is the original EAS update id: manifests keep the id
	// devices already know, so a migrated fleet does not re-download content
	// it is already running.
	storedMetadata := types.UpdateStoredMetadata{
		Platform:   platform,
		CommitHash: historyUpdate.GitCommitHash,
		UpdateUUID: historyUpdate.Id,
		Message:    historyUpdate.Message,
	}
	storedMetadataBytes, err := json.Marshal(storedMetadata)
	if err != nil {
		return err
	}
	if err := s.updateStore.PutFile(ctx, update, "update-metadata.json", bytes.NewReader(storedMetadataBytes)); err != nil {
		return fmt.Errorf("failed to write update-metadata.json into the bucket: %w", err)
	}
	return nil
}

var historyAssetKeyPattern = regexp.MustCompile(`[^a-zA-Z0-9._-]`)

func historyAssetFileName(asset expo.HistoryAsset, index int) string {
	name := historyAssetKeyPattern.ReplaceAllString(asset.Key, "")
	if strings.Trim(name, "._-") == "" {
		return fmt.Sprintf("asset-%d", index)
	}
	return name
}

func (s *Service) invalidateHistoryServingCaches(appId string, touched map[branchRuntime]bool) {
	cache := cache2.GetCache()
	keys := []string{
		dashboard.ComputeGetBranchesCacheKey(appId),
		dashboard.ComputeGetChannelsCacheKey(appId),
	}
	for entry := range touched {
		keys = append(keys,
			dashboard.ComputeGetRuntimeVersionsCacheKey(appId, entry.branch),
			update2.ComputeLastUpdateCacheKey(appId, entry.branch, entry.runtime, types.PlatformIOS),
			update2.ComputeLastUpdateCacheKey(appId, entry.branch, entry.runtime, types.PlatformAndroid),
		)
	}
	for _, key := range keys {
		cache.Delete(key)
	}
}

func (s *Service) deleteHistoryUpdateFolder(ctx context.Context, update types.Update) {
	// The cleanup often follows a canceled job; it must still run.
	if err := s.updateStore.Delete(context.WithoutCancel(ctx), update.AppId, update.Branch, update.RuntimeVersion, update.UpdateId); err != nil {
		log.Printf("[expo-import] failed to clean up update folder %s: %v", update.UpdateId, err)
	}
}

func parseHistoryPlatform(platform string) (types.Platform, bool) {
	switch strings.ToLower(platform) {
	case string(types.PlatformIOS):
		return types.PlatformIOS, true
	case string(types.PlatformAndroid):
		return types.PlatformAndroid, true
	default:
		return "", false
	}
}

func historyPlatformDigit(platform types.Platform) int64 {
	if platform == types.PlatformIOS {
		return 1
	}
	return 2
}