package workers

import (
	"errors"
	"fmt"
	"strconv"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promauto"
	"go.uber.org/zap"
	"gorm.io/gorm"

	"github.com/pingcap/dbaas/billing2/config"
	"github.com/pingcap/dbaas/billing2/constant"
	"github.com/pingcap/dbaas/billing2/db"
	"github.com/pingcap/dbaas/billing2/db/entity"
	"github.com/pingcap/dbaas/billing2/forms"
	"github.com/pingcap/dbaas/billing2/model/billing"
	"github.com/pingcap/dbaas/billing2/utils/log"
	"github.com/pingcap/dbaas/billing2/utils/timeutils"
	"gorm.io/gorm/clause"
)

// Agentic Plan V2 settlement worker (design doc §4/§7:
// https://pingcap.feishu.cn/wiki/GHt2wTPPuiG2Y9k55F2cTIjDn3b).
//
// The worker owns only the schedule: fixed cadence + contiguous
// catch-up, then it enqueues two placeholder derived events per hour
// (agentic-plan:active-fee:per-hour / agentic-plan:idle-fee:per-hour, cost=1 — real
// counts are not known yet). The Stage-2 plug
// (plugs.AgenticPlanUsageAdjuster) computes the real counts inside
// the pricing transaction (compute-only: clamp evidence persists in
// the pricing event's Detail map). A missing H+1 tier1 report fails
// pricing there (ErrAgenticNotReady → price_failed → the retry
// worker's cooldown/deadletter ladder).
//
// The settle lag (now−4h) absorbs the P99 metering latency of the
// tier1 report, so the plug's not-ready error is the tail/outage
// path, not the normal one.
//
// Catch-up semantics with the oneconfig whitelist (no bindings
// table): hours strictly older than the last enqueued hour are never
// revisited. A 2h+ gap is classified by its own billing evidence
// (agenticGapLegacyBilled): a remove→re-add gap was tier1-billed on
// the legacy path and is never re-settled (fresh start from the
// horizon); a worker outage backs up fully — no cap, no manual
// config. A brand-new tenant settles from now−4h only — no backfill.
//
// Operator prerequisites when adding a tenant: the oneconfig
// agentic-plan-tenant-list entry (the entry itself routes the three
// agentic metering families to the new pipeline — no outbox
// whitelist needed), unit price configs for the two derived SKUs and
// tenant_usages rows routing them to usage_type=agentic_plan.

var agenticPlanSettlementsTotal = promauto.NewCounterVec(prometheus.CounterOpts{
	Namespace: "dbaas",
	Subsystem: "billing",
	Name:      "agentic_plan_settlements_total",
	Help: "Agentic plan hourly settlements by result: enqueued / error / " +
		"fresh_start (gap was legacy-billed, skipped)",
}, []string{"result"})

// agenticPlanDelayHours is the settle lag: H = now − 4h. 4h absorbs
// P99 metering latency while staying inside the PRD's billing
// tolerance (1h confirmation delay allowed).
const agenticPlanDelayHours = 4

// SendAgenticPlanMetering is the scheduled entry point.
func SendAgenticPlanMetering() error {
	tenants := agenticTenantIDs()
	if len(tenants) == 0 {
		return nil
	}

	maxHour := timeutils.GetHourDate(timeutils.GetNow().Add(-time.Duration(agenticPlanDelayHours) * time.Hour))
	measureModel := &billing.MeasureModel{}

	for _, tenantID := range tenants {
		meterTenantAgenticPlan(measureModel, tenantID, maxHour)
	}
	return nil
}

// meterTenantAgenticPlan advances the tenant's unsettled hours
// contiguously until caught up.
func meterTenantAgenticPlan(measureModel *billing.MeasureModel, tenantID uint, maxHour time.Time) {
	// Contiguous advance only: from = last enqueued+1h. New tenant
	// (no settlement yet) starts at the settle horizon — no backfill.
	from := maxHour
	if last, ok := agenticLastEnqueuedHour(tenantID); ok {
		from = timeutils.GetHourDate(last.Add(time.Hour))
	}
	if from.After(maxHour) {
		return
	}
	// A gap of 2+ hours means the worker fell behind or the tenant
	// was removed and re-added. Classify by the gap's own billing
	// evidence: a remove→re-add gap was tier1-billed on the legacy
	// path — settling it again would double-charge, so restart from
	// the horizon. A worker outage backs up fully instead.
	if from.Before(maxHour) && agenticGapLegacyBilled(tenantID, from) {
		log.Info("agentic plan: gap was legacy-billed, fresh start",
			zap.Uint("tenantID", tenantID),
			zap.Time("skippedFrom", from), zap.Time("maxHour", maxHour))
		agenticPlanSettlementsTotal.WithLabelValues("fresh_start").Inc()
		from = maxHour
	}

	for hour := from; !hour.After(maxHour); hour = hour.Add(time.Hour) {
		// Placeholder events (cost=1, non-zero to pass pricing
		// validation): deterministic cost IDs make re-enqueues
		// INSERT-IGNORE no-ops on uk_cost_id. The plug replaces the
		// usage with the real counts; zero-count SKUs short-circuit
		// to price_ignore there.
		//
		// CostUnits routes to the outbox enqueue for whitelisted
		// tenants (guarded above) — the standard measure entry, with
		// its metrics and the single routing decision.
		for _, evt := range []*forms.UnitCosts{
			{
				CostID:        agenticActiveCostID(tenantID, hour),
				CostTimeStamp: hour.Unix(),
				Unit:          constant.AgenticPlanActiveFeeSKU,
				Cost:          1,
				Labels:        map[string]string{"metric": "agentic_plan_settlement"},
			},
			{
				CostID:        agenticIdleCostID(tenantID, hour),
				CostTimeStamp: hour.Unix(),
				Unit:          constant.AgenticPlanIdleFeeSKU,
				Cost:          1,
				Labels:        map[string]string{"metric": "agentic_plan_settlement"},
			},
		} {
			if _, err := measureModel.CostUnits(tenantID, evt); err != nil {
				log.Error("agentic plan: enqueue settlement event failed",
					zap.Uint("tenantID", tenantID),
					zap.Time("hour", hour),
					zap.String("unit", evt.Unit), zap.Error(err))
				agenticPlanSettlementsTotal.WithLabelValues("error").Inc()
				return
			}
		}
		// Mark-after-pair: the trigger record is written only after
		// the hour's full event set is enqueued, and the watermark
		// advances with it. A torn enqueue (events in, record not)
		// retries the whole hour next tick — the events' uk_cost_id
		// absorbs the duplicates, and this record's PK absorbs
		// concurrent-pod double marks (INSERT IGNORE).
		record := &entity.WorkerMeteringRecords{
			CostID:        fmt.Sprintf(constant.WorkerMeteringRecordCostIDFormat, constant.AgenticPlanWorkerType, tenantID, hour.Unix()),
			WorkerType:    constant.AgenticPlanWorkerType,
			TenantID:      tenantID,
			CostTimeStamp: hour,
		}
		if err := db.GetDB().Clauses(clause.Insert{Modifier: "IGNORE"}).Create(record).Error; err != nil {
			log.Error("agentic plan: write trigger record failed",
				zap.Uint("tenantID", tenantID),
				zap.Time("hour", hour),
				zap.String("costID", record.CostID), zap.Error(err))
			agenticPlanSettlementsTotal.WithLabelValues("error").Inc()
			return
		}
		agenticPlanSettlementsTotal.WithLabelValues("enqueued").Inc()
	}
}

// ---------------------------------------------------------------------------
// Settlement schedule helpers (formerly the billing/agentic package).
// The pricing-side halves live with their consumers: the per-feature
// routing lane in measure.go (routesAgenticUnit), the aggregation and
// journal in plugs/usage_agentic.go.
// ---------------------------------------------------------------------------

// agenticTenantIDs returns the configured whitelist tenants (oneconfig
// agentic-plan-tenant-list; entries are decimal strings of the tenant
// ID).
func agenticTenantIDs() []uint {
	entries := config.Get().GetStringSlice(config.AgenticPlanTenantList)
	tenants := make([]uint, 0, len(entries))
	for _, e := range entries {
		id, err := strconv.ParseUint(e, 10, 64)
		if err != nil {
			log.Warn("agentic plan: whitelist entry is not a tenant ID",
				zap.String("entry", e), zap.Error(err))
			continue
		}
		tenants = append(tenants, uint(id))
	}
	return tenants
}

// agenticActiveCostID / agenticIdleCostID derive the deterministic
// cost IDs of one hour's settlement events. The format constant is the
// data contract shared with the usage plug's journal — the two sides
// must never drift.
func agenticActiveCostID(tenantID uint, hour time.Time) string {
	return fmt.Sprintf(constant.AgenticPlanActiveCostIDFormat, tenantID, hour.Unix())
}

func agenticIdleCostID(tenantID uint, hour time.Time) string {
	return fmt.Sprintf(constant.AgenticPlanIdleCostIDFormat, tenantID, hour.Unix())
}

// agenticLastEnqueuedHour returns the tenant's most recently handed-off
// trigger hour from worker_metering_records (worker_type=agentic_plan).
// The catch-up loop advances from last+1h only, so hours strictly older
// are never revisited (this is what makes the remove→re-add gap safe).
//
// The record marks the hand-off point, not billing success — pricing
// retries/deadletters stay the pipeline's business. Reads are a
// one-row seek on idx_worker_tenant_time: the old metering_events MAX
// had no (tenant_id, unit) index and walked a big tenant's full event
// history backwards on every tick.
func agenticLastEnqueuedHour(tenantID uint) (time.Time, bool) {
	var record entity.WorkerMeteringRecords
	err := db.GetDB().
		Where("worker_type = ? AND tenant_id = ?", constant.AgenticPlanWorkerType, tenantID).
		Order("cost_time_stamp DESC").First(&record).Error
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return time.Time{}, false
		}
		log.Warn("agentic plan: query last enqueued hour failed",
			zap.Uint("tenantID", tenantID), zap.Error(err))
		return time.Time{}, false
	}
	return timeutils.GetHourDate(record.CostTimeStamp.UTC()), true
}

// agenticGapLegacyBilled reports whether the tenant's first unsettled
// hour was billed on the legacy tier1 path — the signature of a
// remove→re-add gap: while the tenant was off the whitelist, its tier1
// capacity reports priced normally; while on the plan they are
// suppressed to price_ignore. Probes the newest tier1 report in the
// hour's [H+1, H+2) window (the same window the settlement reads).
//
// No report at all carries no legacy-billing evidence (worker outage
// with upstream data missing too) — treat as on-plan and backfill;
// such hours fail pricing as not-ready if their data never arrives.
func agenticGapLegacyBilled(tenantID uint, hour time.Time) bool {
	var evt entity.MeteringEvent
	err := db.GetDB().
		Where("tenant_id = ? AND unit = ?", tenantID, constant.AgenticPlanCapacityUnit).
		Where("cost_timestamp >= ? AND cost_timestamp < ?",
			hour.Add(time.Hour).Unix(), hour.Add(2*time.Hour).Unix()).
		Order("id DESC").First(&evt).Error
	if err != nil {
		if !errors.Is(err, gorm.ErrRecordNotFound) {
			log.Warn("agentic plan: gap classification query failed",
				zap.Uint("tenantID", tenantID), zap.Error(err))
		}
		return false
	}
	return evt.Status != entity.MeteringStatusPriceIgnore
}