package sitefeed

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"log/slog"
	"sync"
	"time"

	"tangled.org/core/appview/cache"
	"tangled.org/core/appview/cloudflare"
	"tangled.org/core/appview/config"
	"tangled.org/core/appview/db"
	"tangled.org/core/appview/models"
	"tangled.org/core/appview/sites"
	"tangled.org/core/hostutil"
	"tangled.org/core/knotfeed"
	"tangled.org/core/log"
	"tangled.org/core/orm"

	"github.com/bluesky-social/indigo/atproto/syntax"
	"github.com/go-git/go-git/v5/plumbing"
	"github.com/redis/go-redis/v9"
)

const (
	registrationRetry    = 30 * time.Second
	maxConcurrentDeploys = 4
)

type feedSource struct {
	cancel context.CancelFunc
}

type Feed struct {
	d      *db.DB
	rdb    *cache.Cache
	cfg    *config.Config
	cf     *cloudflare.Client
	logger *slog.Logger

	deploy func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error

	deploySlots chan struct{}

	deployLocks sync.Map
	pendingSha  sync.Map

	mu      sync.Mutex
	sources map[string]*feedSource
}

func New(d *db.DB, rdb *cache.Cache, cfg *config.Config, cf *cloudflare.Client, logger *slog.Logger) *Feed {
	return &Feed{
		d:           d,
		rdb:         rdb,
		cfg:         cfg,
		cf:          cf,
		logger:      log.SubLogger(logger, "sitefeed"),
		deploy:      sites.Deploy,
		deploySlots: make(chan struct{}, maxConcurrentDeploys),
		sources:     make(map[string]*feedSource),
	}
}

func (f *Feed) Start(ctx context.Context) {
	for {
		knots, err := db.GetRegistrations(f.d, orm.FilterIsNot("registered", "null"))
		if err == nil {
			for _, k := range knots {
				f.Subscribe(ctx, k.Domain)
			}
			return
		}
		f.logger.Error("failed to list registered knots, retrying", "err", err)
		select {
		case <-ctx.Done():
			return
		case <-time.After(registrationRetry):
		}
	}
}

func (f *Feed) Subscribe(ctx context.Context, domain string) {
	host, noTLS, err := hostutil.ParseHostname(domain)
	if err != nil {
		f.logger.Warn("unparseable knot domain, not subscribing", "domain", domain, "err", err)
		return
	}

	f.mu.Lock()
	if _, ok := f.sources[host]; ok {
		f.mu.Unlock()
		return
	}
	srcCtx, cancel := context.WithCancel(ctx)
	src := &feedSource{cancel: cancel}
	f.sources[host] = src
	f.mu.Unlock()

	go func() {
		defer f.forget(host, src)
		f.consumer(host, noTLS).Run(srcCtx)
	}()
}

func (f *Feed) Unsubscribe(domain string) {
	host, _, err := hostutil.ParseHostname(domain)
	if err != nil {
		f.logger.Warn("unparseable knot domain, not unsubscribing", "domain", domain, "err", err)
		return
	}

	f.mu.Lock()
	src, ok := f.sources[host]
	delete(f.sources, host)
	f.mu.Unlock()
	if ok {
		f.logger.Info("unsubscribed from knot firehose", "host", host)
		src.cancel()
	}
}

func (f *Feed) forget(host string, src *feedSource) {
	f.mu.Lock()
	defer f.mu.Unlock()
	if f.sources[host] == src {
		delete(f.sources, host)
	}
}

func (f *Feed) consumer(host string, noTLS bool) *knotfeed.Consumer {
	return &knotfeed.Consumer{
		Host:   host,
		NoTLS:  noTLS,
		Logger: f.logger,

		LoadCursor:  f.loadCursor(host),
		StoreCursor: f.storeCursor(host),
		Handle:      f.handle(host),
		OutdatedReplay: func(_ context.Context, feed knotfeed.Feed) knotfeed.Cursor {
			f.logger.Warn("site feed cursor is behind the knot, resuming live", "host", host, "feed", feed)
			return feed.Live(time.Now())
		},
	}
}

func (f *Feed) loadCursor(host string) func(context.Context) (knotfeed.Cursor, error) {
	return func(ctx context.Context) (knotfeed.Cursor, error) {
		if f.rdb == nil {
			return knotfeed.Cursor{}, nil
		}
		raw, err := f.rdb.Get(ctx, f.cursorKey(host)).Result()
		if errors.Is(err, redis.Nil) {
			return knotfeed.Cursor{}, nil
		}
		if err != nil {
			return knotfeed.Cursor{}, fmt.Errorf("loading site feed cursor: %w", err)
		}
		cursor, err := knotfeed.ParseCursor(raw)
		if err != nil {
			f.logger.Warn("unreadable site feed cursor, resuming live", "host", host, "err", err)
			if err := f.rdb.Del(ctx, f.cursorKey(host)).Err(); err != nil {
				return knotfeed.Cursor{}, fmt.Errorf("clearing unreadable site feed cursor: %w", err)
			}
			return knotfeed.Cursor{}, nil
		}
		return cursor, nil
	}
}

func (f *Feed) storeCursor(host string) func(context.Context, knotfeed.Cursor) error {
	return func(ctx context.Context, cursor knotfeed.Cursor) error {
		if f.rdb == nil {
			return nil
		}
		return f.rdb.Set(ctx, f.cursorKey(host), cursor.Encode(), 0).Err()
	}
}

func (f *Feed) cursorKey(host string) string {
	return "sitefeed:cursor:" + host
}

func (f *Feed) handle(host string) func(context.Context, knotfeed.Message) error {
	return func(ctx context.Context, msg knotfeed.Message) error {
		if msg.Type != knotfeed.TypeCommit || msg.Commit == nil {
			return nil
		}
		var errs []error
		for _, op := range msg.Commit.Records {
			if op.Collection != knotfeed.GitRefCollection {
				continue
			}
			errs = append(errs, f.refOp(ctx, host, msg.Commit.Repo, op))
		}
		return errors.Join(errs...)
	}
}

func (f *Feed) refOp(ctx context.Context, host string, repoDid syntax.DID, op knotfeed.RecordOp) error {
	logger := f.logger.With("knot", host, "repo_did", repoDid)

	refname, ok := knotfeed.UnescapeRkey(op.Rkey)
	if !ok {
		logger.Warn("undecodable git ref rkey", "rkey", op.Rkey)
		return nil
	}

	if op.Deleted() {
		return nil
	}

	record, err := knotfeed.DecodeRefRecord(op.Bytes)
	if err != nil {
		logger.Warn("undecodable git ref record", "ref", refname, "err", err)
		return nil
	}

	repo, err := db.GetRepoByDid(f.d, repoDid.String())
	if errors.Is(err, sql.ErrNoRows) {
		return nil
	}
	if err != nil {
		return fmt.Errorf("looking up repo %s: %w", repoDid, err)
	}
	repoKnot, _, err := hostutil.ParseHostname(repo.Knot)
	if err != nil {
		logger.Warn("repo names an unparseable knot, dropping the record", "repo_knot", repo.Knot)
		return nil
	}
	if repoKnot != host {
		logger.Info("dropping a ref record from a knot that doesn't host the repo", "repo_knot", repo.Knot)
		return nil
	}

	ref := plumbing.ReferenceName(refname)
	if ref.IsBranch() {
		return f.maybeDeploy(ctx, repo, ref.Short(), record.Sha)
	}
	return nil
}

func (f *Feed) maybeDeploy(ctx context.Context, repo *models.Repo, branch string, sha knotfeed.ObjectID) error {
	if f.cf == nil || !f.cf.Enabled() {
		return nil
	}
	siteConfig, err := db.GetRepoSiteConfig(f.d, repo.RepoDid)
	if err != nil {
		return fmt.Errorf("reading the site config for %s: %w", repo.RepoDid, err)
	}
	if siteConfig == nil {
		return nil
	}
	if siteConfig.Branch != branch {
		return nil
	}
	f.pendingSha.Store(repo.RepoDid, sha)
	select {
	case f.deploySlots <- struct{}{}:
	default:
		f.logger.Warn("deploy queue saturated, dropping the deploy", "repo", repo.RepoDid, "branch", branch)
		f.recordDroppedDeploy(repo, siteConfig, sha.String(), "deploy queue saturated")
		return nil
	}
	go func() {
		defer func() { <-f.deploySlots }()
		mu, _ := f.deployLocks.LoadOrStore(repo.RepoDid, &sync.Mutex{})
		mu.(*sync.Mutex).Lock()
		defer mu.(*sync.Mutex).Unlock()
		if latest, ok := f.pendingSha.Load(repo.RepoDid); !ok || latest.(knotfeed.ObjectID) != sha {
			f.logger.Info("superseded push deploy skipped", "repo", repo.RepoDid, "branch", branch, "sha", sha)
			return
		}
		f.triggerDeploy(context.WithoutCancel(ctx), repo, siteConfig, sha.String())
	}()
	return nil
}

func (f *Feed) recordDroppedDeploy(repo *models.Repo, siteConfig *models.RepoSite, sha, reason string) {
	deploy := &models.SiteDeploy{
		RepoDid:   syntax.DID(repo.RepoDid),
		Branch:    siteConfig.Branch,
		Dir:       siteConfig.Dir,
		CommitSHA: sha,
		Trigger:   models.SiteDeployTriggerPush,
		Status:    models.SiteDeployStatusFailure,
		Error:     reason,
	}
	if err := db.AddSiteDeploy(f.d, deploy); err != nil {
		f.logger.Error("failed to record the dropped deploy", "repo", repo.RepoDid, "err", err)
	}
}

func (f *Feed) triggerDeploy(ctx context.Context, repo *models.Repo, siteConfig *models.RepoSite, sha string) {
	logger := f.logger.With("repo", repo.RepoIdentifier())

	deploy := &models.SiteDeploy{
		RepoDid:   syntax.DID(repo.RepoDid),
		Branch:    siteConfig.Branch,
		Dir:       siteConfig.Dir,
		CommitSHA: sha,
		Trigger:   models.SiteDeployTriggerPush,
	}

	deployErr := f.deploy(ctx, f.cf, f.cfg, repo, siteConfig.Branch, siteConfig.Dir)
	if deployErr != nil {
		logger.Error("sites: R2 sync failed on push", "err", deployErr)
		deploy.Status = models.SiteDeployStatusFailure
		deploy.Error = deployErr.Error()
	} else {
		deploy.Status = models.SiteDeployStatusSuccess
	}

	if err := db.AddSiteDeploy(f.d, deploy); err != nil {
		logger.Error("sites: failed to record deploy", "err", err)
	}

	if deployErr == nil {
		logger.Info("site deployed to r2")
	}
}