package worker

import (
	"context"
	cryptorand "crypto/rand"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"regexp"
	"strings"
	"sync"
	"time"

	"tangled.org/core/log"
	"tangled.org/core/migrator/config"
	"tangled.org/core/migrator/crypto"
	"tangled.org/core/migrator/db"
	"tangled.org/core/migrator/git"
	migratoroauth "tangled.org/core/migrator/oauth"
	"tangled.org/core/netutil"
)

const MaxAttempts = 5

var jobDirPattern = regexp.MustCompile(`^job-[0-9]+$`)

var errImportFailed = errors.New("the knot couldn't import the repository from its source")

const (
	partialImportMessage = "the repository was imported, but its Git LFS files weren't, so files tracked with Git LFS are missing on the knot"
	unsourcedRepoMessage = "the knot has no import record for this repository, so the migrator can't tell whether its content arrived"
)

type KnotClient interface {
	CreateRepo(ctx context.Context, ownerDid, knotDid, name, description, sourceURL string) (string, error)
	ImportStatus(ctx context.Context, knotDid, repoDid string) (string, error)
	PutRepoRecord(ctx context.Context, ownerDid, name, repoDid string) error
}

type WorkerPool struct {
	db           *db.DB
	cfg          *config.Config
	proxy        *netutil.SafeConnectProxy
	cmdRunner    git.CommandRunner
	knot         KnotClient
	logger       *slog.Logger
	concurrency  int
	pollInterval time.Duration
	wg           sync.WaitGroup

	githubAPIBase string
}

func NewWorkerPool(
	database *db.DB,
	cfg *config.Config,
	proxyServer *netutil.SafeConnectProxy,
	cmdRunner git.CommandRunner,
	knotClient KnotClient,
	logger *slog.Logger,
) (*WorkerPool, error) {
	if cmdRunner == nil {
		cmdRunner = git.RealCommandRunner{MaxDiskBytes: cfg.MaxDiskBytes}
	}
	if knotClient == nil {
		return nil, errors.New("knot client is required")
	}
	return &WorkerPool{
		db:           database,
		cfg:          cfg,
		proxy:        proxyServer,
		cmdRunner:    cmdRunner,
		knot:         knotClient,
		logger:       log.SubLogger(logger, "worker"),
		concurrency:  cfg.Concurrency,
		pollInterval: 2 * time.Second,
	}, nil
}

func (p *WorkerPool) Start(ctx context.Context) {
	if err := p.startupRecovery(ctx); err != nil {
		p.logger.Error("startup recovery encountered error", "err", err)
	}
	p.logger.Info("starting worker pool", "concurrency", p.concurrency)
	for i := 0; i < p.concurrency; i++ {
		p.wg.Add(1)
		go func(workerID int) {
			defer p.wg.Done()
			p.workerLoop(ctx, workerID)
		}(i)
	}
}

func (p *WorkerPool) Wait() { p.wg.Wait() }

func (p *WorkerPool) startupRecovery(ctx context.Context) error {
	if err := os.MkdirAll(p.cfg.WorkDir, 0700); err != nil {
		return fmt.Errorf("creating work dir %q: %w", p.cfg.WorkDir, err)
	}
	entries, err := os.ReadDir(p.cfg.WorkDir)
	if err == nil {
		for _, entry := range entries {
			if jobDirPattern.MatchString(entry.Name()) {
				_ = os.RemoveAll(filepath.Join(p.cfg.WorkDir, entry.Name()))
			}
		}
	}
	resetCount, err := p.db.ResetInflightJobsToQueued(ctx)
	if err != nil {
		return fmt.Errorf("resetting inflight jobs: %w", err)
	}
	if resetCount > 0 {
		p.logger.Info("recovered inflight jobs to queued", "count", resetCount)
	}
	return nil
}

func (p *WorkerPool) workerLoop(ctx context.Context, id int) {
	for {
		select {
		case <-ctx.Done():
			return
		default:
		}
		job, batch, err := p.db.ClaimNextQueuedJob(ctx)
		if err != nil {
			p.logger.Error("error claiming job", "worker", id, "err", err)
			if !wait(ctx, time.Second) {
				return
			}
			continue
		}
		if job == nil {
			if !wait(ctx, 500*time.Millisecond) {
				return
			}
			continue
		}
		p.executeJob(ctx, job, batch)
	}
}

func wait(ctx context.Context, duration time.Duration) bool {
	timer := time.NewTimer(duration)
	defer timer.Stop()
	select {
	case <-ctx.Done():
		return false
	case <-timer.C:
		return true
	}
}

func (p *WorkerPool) executeJob(ctx context.Context, job *db.Job, batch *db.Batch) {
	jobLogger := p.logger.With("jobId", job.ID, "batchId", batch.ID)
	jobCtx, cancel := context.WithTimeout(ctx, p.cfg.JobTimeout)
	defer cancel()

	var githubToken string
	if job.Private {
		if batch.EncryptedToken == nil || *batch.EncryptedToken == "" {
			p.terminal(ctx, job, batch, db.StatusAuthorizationRequired, strPtr("missing required GitHub credential for private repository"))
			return
		}
		aad := crypto.ComputeAAD(batch.OwnerDid, batch.ID, batch.RequestID)
		token, err := crypto.Decrypt(p.cfg.ParsedMasterKey, *batch.EncryptedToken, aad)
		if err != nil {
			errStr := "failed to decrypt GitHub credential"
			jobLogger.Error(errStr, "err", err)
			p.terminal(ctx, job, batch, db.StatusFailed, &errStr)
			return
		}
		githubToken = token
	}

	capability := ""
	if job.CapabilityToken != nil {
		capability = *job.CapabilityToken
	}
	if job.RepoDid != "" {
		if err := p.db.UpdateJobStatus(ctx, job.ID, db.StatusImporting, nil); err != nil {
			p.handleJobFailure(ctx, job, batch, err, "resuming", githubToken, capability)
			return
		}
		status, err := p.knot.ImportStatus(jobCtx, job.KnotDid, job.RepoDid)
		if err != nil {
			p.handleJobFailure(ctx, job, batch, err, "resuming", githubToken, capability)
			return
		}
		if status != migratoroauth.ImportFailed {
			p.settleImport(jobCtx, ctx, job, batch, status, githubToken, capability)
			return
		}
	}

	// credentials must never reach the knot; private repositories mirror locally
	source := job.SourceURL
	if job.Private {
		jobDir := filepath.Join(p.cfg.WorkDir, fmt.Sprintf("job-%d", job.ID))
		mirrorDir := filepath.Join(jobDir, "repo.git")
		marker := filepath.Join(jobDir, "mirror-ready")
		if _, err := os.Stat(marker); err != nil {
			_ = os.RemoveAll(jobDir)
			if err := os.MkdirAll(jobDir, 0700); err != nil {
				errStr := fmt.Sprintf("creating scratch dir: %v", err)
				p.terminal(ctx, job, batch, db.StatusFailed, &errStr)
				return
			}
			if err := p.probeSourceSize(jobCtx, job, githubToken); err != nil {
				p.handleJobFailure(ctx, job, batch, err, "probing", githubToken, capability)
				return
			}
			if err := p.clonePhase(jobCtx, job, mirrorDir, jobDir, githubToken); err != nil {
				p.handleJobFailure(ctx, job, batch, err, "cloning", githubToken, capability)
				return
			}
			if err := os.WriteFile(marker, nil, 0600); err != nil {
				p.handleJobFailure(ctx, job, batch, err, "cloning", githubToken, capability)
				return
			}
		}
	} else if err := p.probePublicSource(jobCtx, job); err != nil {
		p.handleJobFailure(ctx, job, batch, err, "probing", githubToken, capability)
		return
	}

	if err := p.db.UpdateJobStatus(ctx, job.ID, db.StatusImporting, nil); err != nil {
		p.handleJobFailure(ctx, job, batch, err, "publishing", githubToken, capability)
		return
	}

	// SetJobCapability requires status importing; keep after status update
	if job.Private {
		if capability == "" {
			newCapability, err := newCapabilityToken()
			if err != nil {
				p.handleJobFailure(ctx, job, batch, err, "publishing", githubToken, capability)
				return
			}
			capability = newCapability
			if err := p.db.SetJobCapability(ctx, job.ID, capability); err != nil {
				p.handleJobFailure(ctx, job, batch, err, "publishing", githubToken, capability)
				return
			}
		}
		source = fmt.Sprintf("%s/git/%s/%s.git", strings.TrimSuffix(p.cfg.ServiceURL(), "/"), capability, job.Name)
	}

	if job.RepoDid == "" && p.settleExistingRepo(ctx, jobCtx, job, batch, githubToken, capability) {
		return
	}

	repoDid, err := p.knot.CreateRepo(jobCtx, batch.OwnerDid, job.KnotDid, job.Name, job.Description, source)
	if err != nil {
		if errors.Is(err, migratoroauth.ErrRepoExists) && p.settleConflictRepo(ctx, jobCtx, job, batch, githubToken, capability) {
			return
		}
		p.handleJobFailure(ctx, job, batch, err, "creating", githubToken, capability)
		return
	}
	if err := p.db.SetJobRepoDid(ctx, job.ID, repoDid); err != nil {
		p.handleJobFailure(ctx, job, batch, err, "recording", githubToken, capability)
		return
	}
	job.RepoDid = repoDid

	p.settleImport(jobCtx, ctx, job, batch, migratoroauth.ImportFetching, githubToken, capability)
}

func (p *WorkerPool) adoptableRepoDid(ctx context.Context, ownerDid string, job *db.Job) (string, bool, error) {
	attempts, err := p.db.EarlierRepoAttemptsFor(ctx, ownerDid, job.KnotDid, job.Name)
	if err != nil {
		return "", false, err
	}
	for _, attempt := range attempts {
		if db.SameSource(attempt.SourceURL, job.SourceURL) {
			return attempt.RepoDid, true, nil
		}
	}
	return "", false, nil
}

func (p *WorkerPool) settleExistingRepo(ctx, jobCtx context.Context, job *db.Job, batch *db.Batch, githubToken, capability string) bool {
	repoDid := job.RepoDid
	if repoDid == "" {
		found := false
		var err error
		repoDid, found, err = p.adoptableRepoDid(ctx, batch.OwnerDid, job)
		if err != nil || !found {
			return false
		}
	}
	status, err := p.knot.ImportStatus(jobCtx, job.KnotDid, repoDid)
	if err != nil {
		p.handleJobFailure(ctx, job, batch, err, "checking", githubToken, capability)
		return true
	}
	if status == migratoroauth.ImportFailed {
		return false
	}
	if err := p.db.SetJobRepoDid(ctx, job.ID, repoDid); err != nil {
		p.handleJobFailure(ctx, job, batch, err, "recording", githubToken, capability)
		return true
	}
	job.RepoDid = repoDid
	p.settleImport(jobCtx, ctx, job, batch, status, githubToken, capability)
	return true
}

// resolves 409 on repo creation by adopting an earlier attempt or failing on foreign repos
func (p *WorkerPool) settleConflictRepo(ctx, jobCtx context.Context, job *db.Job, batch *db.Batch, githubToken, capability string) bool {
	repoDid := job.RepoDid
	if repoDid == "" {
		found := false
		var err error
		repoDid, found, err = p.adoptableRepoDid(ctx, batch.OwnerDid, job)
		if err != nil {
			return false
		}
		if !found {
			msg := fmt.Sprintf("a repository named %q already exists on %s and was not created from %s", job.Name, job.KnotDid, job.SourceURL)
			p.terminal(ctx, job, batch, db.StatusFailed, strPtr(msg))
			return true
		}
	}
	status, err := p.knot.ImportStatus(jobCtx, job.KnotDid, repoDid)
	if err != nil {
		return false
	}
	if err := p.db.SetJobRepoDid(ctx, job.ID, repoDid); err != nil {
		p.handleJobFailure(ctx, job, batch, err, "recording", githubToken, capability)
		return true
	}
	job.RepoDid = repoDid
	p.settleImport(jobCtx, ctx, job, batch, status, githubToken, capability)
	return true
}

func (p *WorkerPool) settleImport(jobCtx, ctx context.Context, job *db.Job, batch *db.Batch, status, githubToken, capability string) {
	if status == migratoroauth.ImportFetching {
		var err error
		if status, err = p.waitForImport(jobCtx, job.KnotDid, job.RepoDid); err != nil {
			p.handleJobFailure(ctx, job, batch, err, "waiting", githubToken, capability)
			return
		}
	}
	switch status {
	case migratoroauth.ImportPresent:
		p.recordCompleted(jobCtx, ctx, job, batch, githubToken, capability)
	case migratoroauth.ImportPartial:
		if err := p.knot.PutRepoRecord(jobCtx, batch.OwnerDid, job.Name, job.RepoDid); err != nil {
			p.handleJobFailure(ctx, job, batch, err, "recording", githubToken, capability)
			return
		}
		p.terminal(ctx, job, batch, db.StatusFailed, strPtr(partialImportMessage))
	case migratoroauth.ImportFailed:
		p.handleJobFailure(ctx, job, batch, errImportFailed, "importing", githubToken, capability)
	case "":
		p.terminal(ctx, job, batch, db.StatusFailed, strPtr(unsourcedRepoMessage))
	default:
		p.handleJobFailure(ctx, job, batch, fmt.Errorf("knot reported unknown import status %q", status), "waiting", githubToken, capability)
	}
}

func (p *WorkerPool) recordCompleted(jobCtx, ctx context.Context, job *db.Job, batch *db.Batch, githubToken, capability string) {
	if err := p.knot.PutRepoRecord(jobCtx, batch.OwnerDid, job.Name, job.RepoDid); err != nil {
		p.handleJobFailure(ctx, job, batch, err, "recording", githubToken, capability)
		return
	}
	p.terminal(ctx, job, batch, db.StatusCompleted, nil)
}

func (p *WorkerPool) terminal(ctx context.Context, job *db.Job, batch *db.Batch, status db.JobStatus, errMsg *string) {
	ctx = context.WithoutCancel(ctx)
	if err := p.db.UpdateJobStatus(ctx, job.ID, status, errMsg); err != nil {
		p.logger.Error("marking job terminal", "jobId", job.ID, "status", status, "err", err)
	}
	if err := p.db.RevokeJobCapability(ctx, job.ID); err != nil {
		p.logger.Error("revoking git capability", "jobId", job.ID, "err", err)
	}
	if err := os.RemoveAll(filepath.Join(p.cfg.WorkDir, fmt.Sprintf("job-%d", job.ID))); err != nil {
		p.logger.Error("removing scratch directory", "jobId", job.ID, "err", err)
	}
	_, _ = p.db.CheckAndScrubBatchToken(ctx, batch.ID)
}

func strPtr(s string) *string { return &s }

func sourceHost(rawURL string) string {
	u, err := url.Parse(rawURL)
	if err != nil {
		return ""
	}
	return u.Hostname()
}

func (p *WorkerPool) probeSourceSize(ctx context.Context, job *db.Job, githubToken string) error {
	if p.cfg.MaxPackBytes <= 0 || sourceHost(job.SourceURL) != p.cfg.GitHubHost {
		return nil
	}
	owner, name, ok := githubOwnerRepo(job.SourceURL)
	if !ok {
		return nil
	}
	base := "https://api." + p.cfg.GitHubHost
	if p.githubAPIBase != "" {
		base = p.githubAPIBase
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, base+"/repos/"+owner+"/"+name, nil)
	if err != nil {
		return nil
	}
	if githubToken != "" {
		req.Header.Set("Authorization", "Bearer "+githubToken)
	}
	resp, err := p.gitHubHTTP().Do(req)
	if err != nil {
		return nil
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return nil
	}
	var payload struct {
		Size int64 `json:"size"` // the API reports kilobytes
	}
	if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&payload); err != nil || payload.Size <= 0 {
		return nil
	}
	if payload.Size*1024 > p.cfg.MaxPackBytes {
		return fmt.Errorf("%w: %s/%s reports %d KiB, above the %d byte budget", git.ErrPackLimitExceeded, owner, name, payload.Size, p.cfg.MaxPackBytes)
	}
	return nil
}

// probe before creation so invalid urls do not leave orphan repos on the knot
func (p *WorkerPool) probePublicSource(ctx context.Context, job *db.Job) error {
	if err := p.probeSourceSize(ctx, job, ""); err != nil {
		return err
	}
	proxyAddr := ""
	if p.proxy != nil {
		proxyAddr = p.proxy.Addr()
	}
	env := git.HardenedGitEnv(proxyAddr, "", p.cfg.WorkDir, "", "")
	output, err := p.cmdRunner.Run(ctx, "", env, "git", "ls-remote", "--heads", job.SourceURL)
	if err != nil {
		return git.ClassifyGitError(output, err)
	}
	return nil
}

func (p *WorkerPool) gitHubHTTP() *http.Client {
	transport := http.DefaultTransport.(*http.Transport).Clone()
	if p.proxy != nil {
		proxyURL, err := url.Parse("http://" + p.proxy.Addr())
		if err != nil {
			transport.Proxy = nil
		} else {
			transport.Proxy = http.ProxyURL(proxyURL)
		}
	}
	return &http.Client{Transport: transport, Timeout: 30 * time.Second}
}

func githubOwnerRepo(rawURL string) (string, string, bool) {
	u, err := url.Parse(rawURL)
	if err != nil {
		return "", "", false
	}
	parts := strings.Split(strings.Trim(u.Path, "/"), "/")
	if len(parts) < 2 {
		return "", "", false
	}
	owner, name := parts[0], strings.TrimSuffix(parts[1], ".git")
	if owner == "" || name == "" {
		return "", "", false
	}
	return owner, name, true
}

func newCapabilityToken() (string, error) {
	raw := make([]byte, 32)
	if _, err := cryptorand.Read(raw); err != nil {
		return "", fmt.Errorf("generating git capability: %w", err)
	}
	return base64.RawURLEncoding.EncodeToString(raw), nil
}

func mirrorPackBytes(mirror string) (int64, error) {
	entries, err := os.ReadDir(filepath.Join(mirror, "objects", "pack"))
	if os.IsNotExist(err) {
		return 0, nil
	}
	if err != nil {
		return 0, err
	}
	var total int64
	for _, entry := range entries {
		if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".pack") {
			continue
		}
		info, err := entry.Info()
		if err != nil {
			return 0, err
		}
		total += info.Size()
	}
	return total, nil
}

func (p *WorkerPool) waitForImport(ctx context.Context, knotDid, repoDid string) (string, error) {
	var lastErr error
	for {
		status, err := p.knot.ImportStatus(ctx, knotDid, repoDid)
		switch {
		case err != nil:
			lastErr = err
		case status != migratoroauth.ImportFetching:
			return status, nil
		default:
			lastErr = nil
		}
		if !wait(ctx, p.pollInterval) {
			if lastErr != nil {
				return "", fmt.Errorf("waiting for the knot import: %w (last answer: %v)", ctx.Err(), lastErr)
			}
			return "", fmt.Errorf("waiting for the knot import: %w", ctx.Err())
		}
	}
}

func (p *WorkerPool) clonePhase(ctx context.Context, job *db.Job, mirrorDir, jobDir, githubToken string) error {
	var askpassBin, askpassHost, askpassToken string
	if job.Private {
		var err error
		askpassBin, err = git.CreateAskpassScript(jobDir)
		if err != nil {
			return fmt.Errorf("creating clone askpass script: %w", err)
		}
		defer os.Remove(askpassBin)
		askpassHost, askpassToken = sourceHost(job.SourceURL), githubToken
	}
	proxyAddr := ""
	if p.proxy != nil {
		proxyAddr = p.proxy.Addr()
	}
	env := git.HardenedGitEnv(proxyAddr, askpassBin, jobDir, askpassHost, askpassToken)
	runner := p.cmdRunner
	if real, ok := runner.(git.RealCommandRunner); ok {
		real.WatchDir = jobDir
		runner = real
	}
	output, err := runner.Run(ctx, jobDir, env, "git", "clone", "--mirror", job.SourceURL, mirrorDir)
	if err != nil {
		return git.ClassifyGitError(output, err)
	}
	if p.cfg.MaxPackBytes > 0 {
		packBytes, err := mirrorPackBytes(mirrorDir)
		if err != nil {
			return fmt.Errorf("measuring cloned pack: %w", err)
		}
		if packBytes > p.cfg.MaxPackBytes {
			return fmt.Errorf("%w: cloned pack is %d bytes, max %d", git.ErrPackLimitExceeded, packBytes, p.cfg.MaxPackBytes)
		}
	}
	if _, err := runner.LookPath("git-lfs"); err == nil {
		output, err := runner.Run(ctx, mirrorDir, env, "git", "lfs", "fetch", "--all")
		if err != nil {
			p.logger.Error("git lfs fetch failed", "out", git.RedactSecrets(output, githubToken), "err", git.RedactSecrets(err.Error(), githubToken))
			return git.ClassifyGitError(output, err)
		}
	}
	return nil
}

func (p *WorkerPool) handleJobFailure(ctx context.Context, job *db.Job, batch *db.Batch, err error, phase, githubToken, capability string) {
	stopping := ctx.Err() != nil
	safeErr := git.RedactSecrets(err.Error(), githubToken, capability)
	p.logger.Error("job failed during phase", "jobId", job.ID, "phase", phase, "attempts", job.Attempts, "err", safeErr)
	ctx = context.WithoutCancel(ctx)
	// refused grants invalidate the refresh-token family; retrying cannot succeed
	if errors.Is(err, git.ErrAuthRequired) || errors.Is(err, migratoroauth.ErrGrantRequired) {
		p.terminal(ctx, job, batch, db.StatusAuthorizationRequired, &safeErr)
		return
	}
	if errors.Is(err, git.ErrDiskLimitExceeded) || errors.Is(err, git.ErrPackLimitExceeded) {
		p.terminal(ctx, job, batch, db.StatusFailed, &safeErr)
		return
	}
	if stopping {
		if err := p.db.ReleaseJob(ctx, job.ID, &safeErr); err != nil {
			p.logger.Error("releasing job after cancellation", "jobId", job.ID, "err", err)
		}
		return
	}
	if job.Attempts < MaxAttempts {
		backoff := min(time.Duration(1<<uint(job.Attempts))*time.Second, 30*time.Second)
		_ = p.db.ScheduleJobRetry(ctx, job.ID, backoff, &safeErr)
		return
	}
	p.terminal(ctx, job, batch, db.StatusFailed, &safeErr)
}