package metrics

import (
	"strings"
	"sync"
	"time"
)

// defaultFileGCDrainHitCapTTL must sit above one FileGC lease (5m) and one
// safety-net scan (5m) so an in-flight batch or a lost kick waiting for the
// scan cannot lift unless while claimable is still exported. It stays below
// the 15m stale-backlog alert window.
const defaultFileGCDrainHitCapTTL = 10 * time.Minute

// fileGCDrainHitCapTTL drops drive9_file_gc_drain_hit_cap that has not been
// refreshed by a FileGC drain. Tenant-wide 15m retention is refreshed by
// ordinary HTTP/DB activity, which would otherwise keep the unless-suppressor
// set after a stalled follow-up kick. Claimable/age are not expired here so a
// leftover backlog can still page once hit_cap is gone.
var fileGCDrainHitCapTTL = defaultFileGCDrainHitCapTTL

// fileGCBacklogClock is the time source for hit-cap last-touch expiry. Tests swap it.
var fileGCBacklogClock = time.Now

type fileGCBacklogTouchKey struct {
	tenantID string
	orgID    string
}

var fileGCBacklogTouch = struct {
	mu   sync.Mutex
	last map[fileGCBacklogTouchKey]time.Time
}{last: make(map[fileGCBacklogTouchKey]time.Time)}

func fileGCBacklogGauges() []*Float64Gauge {
	return []*Float64Gauge{
		fileGCClaimable,
		fileGCOldestClaimableAgeSeconds,
		fileGCDrainHitCap,
		fileGCClaimStalled,
	}
}

func fileGCBacklogGaugeNames() []string {
	gauges := fileGCBacklogGauges()
	names := make([]string, 0, len(gauges))
	for _, g := range gauges {
		if g == nil || g.name == "" {
			continue
		}
		names = append(names, g.name)
	}
	return names
}

func fileGCBacklogAttrs(tenantID, orgID string) (string, string, []Attribute) {
	cleanTenantID := cleanMetricValue(tenantID, "unknown")
	orgID = cleanTiDBCloudOrgID(orgID)
	return cleanTenantID, orgID, []Attribute{
		Attr("tenant_id", cleanTenantID),
		Attr("tidbcloud_org_id", orgID),
	}
}

func touchFileGCDrainHitCap(tenantID, orgID string) {
	if tenantID == "" || tenantID == "unknown" {
		return
	}
	fileGCBacklogTouch.mu.Lock()
	fileGCBacklogTouch.last[fileGCBacklogTouchKey{tenantID: tenantID, orgID: orgID}] = fileGCBacklogClock()
	fileGCBacklogTouch.mu.Unlock()
}

func forgetFileGCDrainHitCapTouch(tenantID, orgID string) {
	fileGCBacklogTouch.mu.Lock()
	delete(fileGCBacklogTouch.last, fileGCBacklogTouchKey{tenantID: tenantID, orgID: orgID})
	fileGCBacklogTouch.mu.Unlock()
}

func forgetFileGCDrainHitCapTouchTenant(tenantID string) {
	if tenantID == "" {
		return
	}
	fileGCBacklogTouch.mu.Lock()
	for k := range fileGCBacklogTouch.last {
		if k.tenantID == tenantID {
			delete(fileGCBacklogTouch.last, k)
		}
	}
	fileGCBacklogTouch.mu.Unlock()
}

func deleteFileGCBacklogGauges(tenantID, orgID string) {
	_, _, attrs := fileGCBacklogAttrs(tenantID, orgID)
	fileGCClaimable.Delete(attrs...)
	fileGCOldestClaimableAgeSeconds.Delete(attrs...)
	fileGCDrainHitCap.Delete(attrs...)
	forgetFileGCDrainHitCapTouch(tenantID, orgID)
}

// ObserveFileGCDrainHitCap records whether the last FileGC drain for this
// tenant filled its batch (more claimable work likely remains). Sparse: the
// series is omitted when false so Drive9FileGCClaimableBacklogStale can
// suppress tenants that are still making drain progress.
func ObserveFileGCDrainHitCap(tenantID, tidbCloudOrgID string, hitCap bool) {
	tenantID = strings.TrimSpace(tenantID)
	if tenantID == "" {
		return
	}
	RegisterModule("file_gc")
	cleanTenantID, orgID, attrs := fileGCBacklogAttrs(tenantID, tidbCloudOrgID)
	if cleanTenantID == "unknown" {
		return
	}
	if !hitCap {
		fileGCDrainHitCap.Delete(attrs...)
		forgetFileGCDrainHitCapTouch(cleanTenantID, orgID)
		return
	}
	fileGCDrainHitCap.Set(1, attrs...)
	touchFileGCDrainHitCap(cleanTenantID, orgID)
}

// TouchFileGCDrainHitCap refreshes hit-cap last-touch without changing the
// value. Used at drain start and after each claimed task so an in-flight
// batch longer than the TTL does not drop the unless-suppressor. No-op when
// this process is not exporting hit_cap for the tenant.
func TouchFileGCDrainHitCap(tenantID, tidbCloudOrgID string) {
	tenantID = strings.TrimSpace(tenantID)
	if tenantID == "" {
		return
	}
	cleanTenantID, orgID, _ := fileGCBacklogAttrs(tenantID, tidbCloudOrgID)
	if cleanTenantID == "unknown" {
		return
	}
	fileGCBacklogTouch.mu.Lock()
	k := fileGCBacklogTouchKey{tenantID: cleanTenantID, orgID: orgID}
	if _, ok := fileGCBacklogTouch.last[k]; ok {
		fileGCBacklogTouch.last[k] = fileGCBacklogClock()
	}
	fileGCBacklogTouch.mu.Unlock()
}

// SweepStaleFileGCDrainHitCap deletes hit-cap and claim_stalled series whose
// last FileGC touch is older than fileGCDrainHitCapTTL. Called from the scrape
// path. Does not delete claimable/age — leftover backlog must remain visible
// after unless lifts.
func SweepStaleFileGCDrainHitCap() {
	ttl := fileGCDrainHitCapTTL
	if ttl <= 0 {
		return
	}
	now := fileGCBacklogClock()
	var stale []fileGCBacklogTouchKey
	fileGCBacklogTouch.mu.Lock()
	for k, ts := range fileGCBacklogTouch.last {
		if now.Sub(ts) >= ttl {
			stale = append(stale, k)
			delete(fileGCBacklogTouch.last, k)
		}
	}
	fileGCBacklogTouch.mu.Unlock()
	for _, k := range stale {
		fileGCBacklogTouch.mu.Lock()
		_, refreshed := fileGCBacklogTouch.last[k]
		fileGCBacklogTouch.mu.Unlock()
		if refreshed {
			// A SET raced in after we dropped the stale timestamp.
			continue
		}
		_, _, attrs := fileGCBacklogAttrs(k.tenantID, k.orgID)
		fileGCDrainHitCap.Delete(attrs...)
		fileGCClaimStalled.Delete(attrs...)
	}
}

// ObserveFileGCClaimStalled records a sparse owner-local signal that FileGC
// claim/COUNT mismatch retries are exhausted while due rows remain. It does
// not revive drive9_file_gc_claimable, so Drive9FileGCClaimableBacklogStale
// stays quiet for SKIP LOCKED stalls.
func ObserveFileGCClaimStalled(tenantID, tidbCloudOrgID string, stalled bool) {
	tenantID = strings.TrimSpace(tenantID)
	if tenantID == "" {
		return
	}
	RegisterModule("file_gc")
	cleanTenantID, orgID, attrs := fileGCBacklogAttrs(tenantID, tidbCloudOrgID)
	if cleanTenantID == "unknown" {
		return
	}
	if !stalled {
		fileGCClaimStalled.Delete(attrs...)
		return
	}
	fileGCClaimStalled.Set(1, attrs...)
	touchFileGCDrainHitCap(cleanTenantID, orgID)
}

// FileGCBacklogTenantIDs lists tenants that currently export any FileGC
// backlog gauge on this process. Used to drop leftover series after shard
// ownership moves off this pod.
func FileGCBacklogTenantIDs() []string {
	return globalRegistry.namedGaugeTenantIDs(fileGCBacklogGaugeNames()...)
}

// ClearFileGCClaimableBacklog drops this process's FileGC backlog gauges for
// tenantID (any org label). Used when the tenant backend leaves this pod or
// this pod loses shard ownership, so a leftover sparse series cannot keep
// paging.
func ClearFileGCClaimableBacklog(tenantID string) {
	tenantID = strings.TrimSpace(tenantID)
	if tenantID == "" {
		return
	}
	cleanTenantID := cleanMetricValue(tenantID, "unknown")
	if cleanTenantID == "unknown" {
		return
	}
	RegisterModule("file_gc")
	for _, name := range fileGCBacklogGaugeNames() {
		globalRegistry.deleteNamedGaugesByTenant(name, cleanTenantID)
	}
	forgetFileGCDrainHitCapTouchTenant(cleanTenantID)
}