package storage

import (
	"context"
	"database/sql"
	"log/slog"
	"strings"
	"time"

	"go.mongodb.org/mongo-driver/v2/bson"
	"go.mongodb.org/mongo-driver/v2/mongo"
	"go.mongodb.org/mongo-driver/v2/mongo/options"

	"dashboard-cdc-worker/internal/dashboard"
)

const activeLogCollection = "user_daily_active_log"

type activeLogRow struct {
	UserID              int64  `bson:"userId"`
	ActiveDay           string `bson:"activeDate"`
	SysOrigin           string `bson:"sysOrigin"`
	RegisterCountryCode string `bson:"registerCountryCode"`
}

type activeLogEvent struct {
	UserID    int64
	SysOrigin string
	Country   dashboard.Country
}

type activeDayStats struct {
	active map[string]map[int64]struct{}
}

type activeRetentionValues struct {
	countryName string
	activeUsers int64
	d1User      int64
	d1Base      int64
	d7User      int64
	d7Base      int64
	d30User     int64
	d30Base     int64
}

type countryCount struct {
	name  string
	count int64
}

func (r *MySQLRepository) StartActiveReconciler(ctx context.Context, db *mongo.Database, logger *slog.Logger) {
	if !r.cfg.Active.Enabled || r.cfg.Active.Interval <= 0 {
		return
	}
	run := func() {
		if err := r.RebuildRecentActiveRetention(ctx, db); err != nil {
			if r.metrics != nil {
				r.metrics.RecordError(err)
			}
			logger.Error("dashboard active reconcile failed", "error", err)
			return
		}
		logger.Info("dashboard active reconcile applied", "lookbackDays", r.cfg.Active.LookbackDays)
	}
	go func() {
		run()
		ticker := time.NewTicker(r.cfg.Active.Interval)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				run()
			}
		}
	}()
}

func (r *MySQLRepository) RebuildRecentActiveRetention(ctx context.Context, db *mongo.Database) error {
	location, err := time.LoadLocation(r.cfg.Dashboard.StorageTimezone)
	if err != nil {
		location = time.UTC
	}
	today := time.Now().In(location)
	end := time.Date(today.Year(), today.Month(), today.Day(), 0, 0, 0, 0, location).AddDate(0, 0, 1)
	start := end.AddDate(0, 0, -r.cfg.Active.LookbackDays)
	return r.RebuildActiveRetentionRange(ctx, db, start, end)
}

func (r *MySQLRepository) RebuildActiveRetentionRange(ctx context.Context, db *mongo.Database, start, end time.Time) error {
	events, err := r.loadActiveLogs(ctx, db, start, end)
	if err != nil {
		return err
	}
	users, err := r.loadUserCountriesByID(ctx, collectActiveUserIDs(events))
	if err != nil {
		return err
	}
	return r.WithDashboardLock(ctx, r.cfg.CDC.BackfillLockWait, func() error {
		return r.WithTx(ctx, func(tx *sql.Tx) error {
			for day := start; day.Before(end); day = day.AddDate(0, 0, 1) {
				if err := r.replaceActiveRetentionDay(ctx, tx, day, events, users); err != nil {
					return err
				}
			}
			for _, statTimezone := range r.cfg.Dashboard.StatTimezones {
				if err := r.rebuildAllPeriodFromDays(ctx, tx, statTimezone); err != nil {
					return err
				}
			}
			return nil
		})
	})
}

func (r *MySQLRepository) loadActiveLogs(ctx context.Context, db *mongo.Database, start, end time.Time) (map[string]map[int64]activeLogEvent, error) {
	filter := bson.D{
		{Key: "activeDate", Value: bson.D{{Key: "$gte", Value: start.Format("2006-01-02")}, {Key: "$lt", Value: end.Format("2006-01-02")}}},
		{Key: "sysOrigin", Value: bson.D{{Key: "$in", Value: r.cfg.Dashboard.SysOrigins}}},
	}
	findOptions := options.Find().SetProjection(bson.D{
		{Key: "userId", Value: 1},
		{Key: "activeDate", Value: 1},
		{Key: "sysOrigin", Value: 1},
		{Key: "registerCountryCode", Value: 1},
	})
	if r.cfg.Active.BatchLimit > 0 {
		findOptions.SetLimit(int64(r.cfg.Active.BatchLimit))
	}
	cursor, err := db.Collection(activeLogCollection).Find(ctx, filter, findOptions)
	if err != nil {
		return nil, err
	}
	defer cursor.Close(ctx)

	events := make(map[string]map[int64]activeLogEvent)
	for cursor.Next(ctx) {
		var row activeLogRow
		if err := cursor.Decode(&row); err != nil {
			return nil, err
		}
		if row.UserID == 0 || row.ActiveDay == "" || !r.allowedOrigin(row.SysOrigin) {
			continue
		}
		users := events[row.ActiveDay]
		if users == nil {
			users = make(map[int64]activeLogEvent)
			events[row.ActiveDay] = users
		}
		users[row.UserID] = activeLogEvent{
			UserID:    row.UserID,
			SysOrigin: strings.TrimSpace(row.SysOrigin),
			Country:   r.activeLogCountry(row.RegisterCountryCode),
		}
	}
	return events, cursor.Err()
}

func (r *MySQLRepository) activeLogCountry(registerCountryCode string) dashboard.Country {
	code := strings.TrimSpace(registerCountryCode)
	if code == "" {
		code = r.cfg.Dashboard.UnknownCountryCode
	}
	return dashboard.Country{Code: code, Name: code}
}

func collectActiveUserIDs(events map[string]map[int64]activeLogEvent) []int64 {
	seen := make(map[int64]struct{})
	for _, users := range events {
		for userID := range users {
			seen[userID] = struct{}{}
		}
	}
	out := make([]int64, 0, len(seen))
	for userID := range seen {
		out = append(out, userID)
	}
	return out
}

func activeUserSet(events map[int64]activeLogEvent) map[int64]struct{} {
	out := make(map[int64]struct{}, len(events))
	for userID := range events {
		out[userID] = struct{}{}
	}
	return out
}

func (r *MySQLRepository) loadUserCountriesByID(ctx context.Context, ids []int64) (map[int64]dashboard.UserCountry, error) {
	result := make(map[int64]dashboard.UserCountry)
	for start := 0; start < len(ids); start += bulkInsertChunkSize {
		end := min(start+bulkInsertChunkSize, len(ids))
		args := make([]interface{}, 0, end-start)
		for _, id := range ids[start:end] {
			args = append(args, id)
		}
		rows, err := r.db.QueryContext(ctx, `
SELECT user_id, sys_origin, country_code, country_name, is_deleted, user_created_at, user_updated_at
FROM dashboard_cdc_user_country
WHERE user_id IN `+inPlaceholders(end-start), args...)
		if err != nil {
			return nil, err
		}
		for rows.Next() {
			var user dashboard.UserCountry
			var createdAt, updatedAt sql.NullTime
			if err := rows.Scan(&user.UserID, &user.SysOrigin, &user.Country.Code, &user.Country.Name,
				&user.IsDeleted, &createdAt, &updatedAt); err != nil {
				_ = rows.Close()
				return nil, err
			}
			if createdAt.Valid {
				user.CreatedAt = createdAt.Time
			}
			if updatedAt.Valid {
				user.UpdatedAt = updatedAt.Time
			}
			result[user.UserID] = user
		}
		if err := rows.Close(); err != nil {
			return nil, err
		}
	}
	return result, nil
}

func (r *MySQLRepository) replaceActiveRetentionDay(ctx context.Context, tx *sql.Tx, day time.Time,
	events map[string]map[int64]activeLogEvent, users map[int64]dashboard.UserCountry) error {
	dayKey := day.Format("2006-01-02")
	statsByOrigin := make(map[string]map[string]*activeDayStats)
	for userID, event := range events[dayKey] {
		sysOrigin := strings.TrimSpace(event.SysOrigin)
		if !r.allowedOrigin(sysOrigin) {
			continue
		}
		country := event.Country
		if country.Code == "" {
			if user, ok := users[userID]; ok && user.Country.Code != "" {
				country = user.Country
			} else {
				country = dashboard.Country{Code: r.cfg.Dashboard.UnknownCountryCode, Name: r.cfg.Dashboard.UnknownCountryCode}
			}
		}
		if country.Name == "" {
			country.Name = country.Code
		}
		stats := statsByOrigin[sysOrigin]
		if stats == nil {
			stats = make(map[string]*activeDayStats)
			statsByOrigin[sysOrigin] = stats
		}
		stat := stats[country.Code]
		if stat == nil {
			stat = &activeDayStats{active: make(map[string]map[int64]struct{})}
			stats[country.Code] = stat
		}
		active := stat.active[country.Name]
		if active == nil {
			active = make(map[int64]struct{})
			stat.active[country.Name] = active
		}
		active[userID] = struct{}{}
	}
	retentionByOrigin, err := r.retentionStatsForDay(ctx, day, activeUserSet(events[dayKey]), users)
	if err != nil {
		return err
	}
	for _, statTimezone := range r.cfg.Dashboard.StatTimezones {
		for _, sysOrigin := range r.cfg.Dashboard.SysOrigins {
			if _, err := tx.ExecContext(ctx, `
UPDATE country_dashboard_period_metric
SET daily_active_user = 0,
    d1_retention_user = 0, d1_retention_base_user = 0,
    d7_retention_user = 0, d7_retention_base_user = 0,
    d30_retention_user = 0, d30_retention_base_user = 0,
    refreshed_at = NOW()
WHERE period_type = 'DAY' AND period_key = ? AND stat_timezone = ? AND sys_origin = ?`, dayKey, statTimezone, sysOrigin); err != nil {
				return err
			}
			if err := r.upsertActiveRetentionRows(ctx, tx, day, statTimezone, sysOrigin,
				statsByOrigin[sysOrigin], retentionByOrigin[sysOrigin]); err != nil {
				return err
			}
		}
	}
	return nil
}

func (r *MySQLRepository) retentionStatsForDay(ctx context.Context, day time.Time,
	activeUsers map[int64]struct{}, users map[int64]dashboard.UserCountry) (map[string]map[string]activeRetentionValues, error) {
	out := make(map[string]map[string]activeRetentionValues)
	for _, days := range []int{1, 7, 30} {
		bases, err := r.loadRetentionBase(ctx, day.AddDate(0, 0, -days))
		if err != nil {
			return nil, err
		}
		for origin, countries := range bases {
			originOut := out[origin]
			if originOut == nil {
				originOut = make(map[string]activeRetentionValues)
				out[origin] = originOut
			}
			for country, item := range countries {
				stats := originOut[country]
				stats.countryName = item.name
				switch days {
				case 1:
					stats.d1Base = item.count
				case 7:
					stats.d7Base = item.count
				case 30:
					stats.d30Base = item.count
				}
				originOut[country] = stats
			}
		}
	}
	for userID := range activeUsers {
		user, ok := users[userID]
		if !ok || user.CreatedAt.IsZero() || user.IsDeleted || !r.allowedOrigin(user.SysOrigin) {
			continue
		}
		for _, days := range []int{1, 7, 30} {
			if user.CreatedAt.Format("2006-01-02") != day.AddDate(0, 0, -days).Format("2006-01-02") {
				continue
			}
			originOut := out[user.SysOrigin]
			if originOut == nil {
				originOut = make(map[string]activeRetentionValues)
				out[user.SysOrigin] = originOut
			}
			country := user.Country.Code
			stats := originOut[country]
			if stats.countryName == "" {
				stats.countryName = user.Country.Name
			}
			switch days {
			case 1:
				stats.d1User++
			case 7:
				stats.d7User++
			case 30:
				stats.d30User++
			}
			originOut[country] = stats
		}
	}
	return out, nil
}

func (r *MySQLRepository) loadRetentionBase(ctx context.Context, cohortDay time.Time) (map[string]map[string]countryCount, error) {
	args := stringArgs(r.cfg.Dashboard.SysOrigins)
	args = append(args, cohortDay, cohortDay.AddDate(0, 0, 1))
	rows, err := r.db.QueryContext(ctx, `
SELECT sys_origin, country_code, MAX(country_name), COUNT(1)
FROM dashboard_cdc_user_country
WHERE is_deleted = 0
  AND sys_origin IN `+inStringPlaceholders(len(r.cfg.Dashboard.SysOrigins))+`
  AND user_created_at >= ?
  AND user_created_at < ?
GROUP BY sys_origin, country_code`, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	result := make(map[string]map[string]countryCount)
	for rows.Next() {
		var origin, country, name string
		var count int64
		if err := rows.Scan(&origin, &country, &name, &count); err != nil {
			return nil, err
		}
		countries := result[origin]
		if countries == nil {
			countries = make(map[string]countryCount)
			result[origin] = countries
		}
		countries[country] = countryCount{name: name, count: count}
	}
	return result, rows.Err()
}

func (r *MySQLRepository) upsertActiveRetentionRows(ctx context.Context, tx *sql.Tx, day time.Time,
	statTimezone string, sysOrigin string, active map[string]*activeDayStats, retention map[string]activeRetentionValues) error {
	values := make(map[string]activeRetentionValues)
	for country, stat := range active {
		item := values[country]
		for name, users := range stat.active {
			item.countryName = name
			item.activeUsers += int64(len(users))
		}
		values[country] = item
	}
	for country, item := range retention {
		existing := values[country]
		if existing.countryName == "" {
			existing.countryName = item.countryName
		}
		existing.d1User = item.d1User
		existing.d1Base = item.d1Base
		existing.d7User = item.d7User
		existing.d7Base = item.d7Base
		existing.d30User = item.d30User
		existing.d30Base = item.d30Base
		values[country] = existing
	}
	if len(values) == 0 {
		return nil
	}
	dayKey := day.Format("2006-01-02")
	args := make([]interface{}, 0, len(values)*16)
	for country, item := range values {
		name := item.countryName
		if name == "" {
			name = country
		}
		args = append(args, dashboard.PeriodDay, dayKey, statTimezone, dayKey, day,
			day.AddDate(0, 0, 1), sysOrigin, country, name, item.activeUsers,
			item.d1User, item.d1Base, item.d7User, item.d7Base, item.d30User, item.d30Base)
	}
	_, err := tx.ExecContext(ctx, `
INSERT INTO country_dashboard_period_metric (
  period_type, period_key, stat_timezone, period_name, period_start_date, period_end_date,
  sys_origin, country_code, country_name, daily_active_user,
  d1_retention_user, d1_retention_base_user,
  d7_retention_user, d7_retention_base_user,
  d30_retention_user, d30_retention_base_user
) VALUES `+placeholders(len(values), 16)+`
ON DUPLICATE KEY UPDATE
  period_name = VALUES(period_name),
  period_start_date = VALUES(period_start_date),
  period_end_date = VALUES(period_end_date),
  country_name = VALUES(country_name),
  daily_active_user = VALUES(daily_active_user),
  d1_retention_user = VALUES(d1_retention_user),
  d1_retention_base_user = VALUES(d1_retention_base_user),
  d7_retention_user = VALUES(d7_retention_user),
  d7_retention_base_user = VALUES(d7_retention_base_user),
  d30_retention_user = VALUES(d30_retention_user),
  d30_retention_base_user = VALUES(d30_retention_base_user),
  refreshed_at = NOW()`, args...)
	return err
}

func (r *MySQLRepository) rebuildAllPeriodFromDays(ctx context.Context, tx *sql.Tx, statTimezone string) error {
	for _, sysOrigin := range r.cfg.Dashboard.SysOrigins {
		if _, err := tx.ExecContext(ctx, `
DELETE FROM country_dashboard_period_user_metric
WHERE period_type = 'ALL' AND period_key = 'ALL' AND stat_timezone = ? AND sys_origin = ?`, statTimezone, sysOrigin); err != nil {
			return err
		}
		if _, err := tx.ExecContext(ctx, `
INSERT IGNORE INTO country_dashboard_period_user_metric (
  period_type, period_key, stat_timezone, period_name, period_start_date, period_end_date,
  sys_origin, country_code, country_name, user_id, metric_type
)
SELECT DISTINCT 'ALL', 'ALL', ?, '全部', NULL, NULL,
  sys_origin, country_code, country_name, user_id, metric_type
FROM country_dashboard_period_user_metric
WHERE period_type = 'DAY' AND stat_timezone = ? AND sys_origin = ?`, statTimezone, statTimezone, sysOrigin); err != nil {
			return err
		}
		if _, err := tx.ExecContext(ctx, `
DELETE FROM country_dashboard_period_metric
WHERE period_type = 'ALL' AND period_key = 'ALL' AND stat_timezone = ? AND sys_origin = ?`, statTimezone, sysOrigin); err != nil {
			return err
		}
		if _, err := tx.ExecContext(ctx, `
INSERT INTO country_dashboard_period_metric (
  period_type, period_key, stat_timezone, period_name, period_start_date, period_end_date,
  sys_origin, country_code, country_name,
  country_new_user, daily_active_user, new_user_recharge, official_recharge, mifapay_recharge,
  google_recharge, dealer_recharge, salary_exchange, salary_transfer, gift_consume,
  lucky_gift_total_flow, lucky_gift_user, lucky_gift_payout, lucky_gift_anchor_share,
  game_total_flow, game_user, game_payout,
  d1_retention_user, d1_retention_base_user, d7_retention_user, d7_retention_base_user,
  d30_retention_user, d30_retention_base_user, refreshed_at
)
SELECT 'ALL', 'ALL', ?, '全部', NULL, NULL,
  d.sys_origin, d.country_code, MAX(d.country_name),
  IFNULL(MAX(u.country_new_user), 0),
  IFNULL(SUM(d.daily_active_user), 0),
  IFNULL(SUM(d.new_user_recharge), 0),
  IFNULL(SUM(d.official_recharge), 0),
  IFNULL(SUM(d.mifapay_recharge), 0),
  IFNULL(SUM(d.google_recharge), 0),
  IFNULL(SUM(d.dealer_recharge), 0),
  IFNULL(SUM(d.salary_exchange), 0),
  IFNULL(SUM(d.salary_transfer), 0),
  IFNULL(SUM(d.gift_consume), 0),
  IFNULL(SUM(d.lucky_gift_total_flow), 0),
  IFNULL(MAX(u.lucky_gift_user), 0),
  IFNULL(SUM(d.lucky_gift_payout), 0),
  IFNULL(SUM(d.lucky_gift_anchor_share), 0),
  IFNULL(SUM(d.game_total_flow), 0),
  IFNULL(MAX(u.game_user), 0),
  IFNULL(SUM(d.game_payout), 0),
  IFNULL(SUM(d.d1_retention_user), 0),
  IFNULL(SUM(d.d1_retention_base_user), 0),
  IFNULL(SUM(d.d7_retention_user), 0),
  IFNULL(SUM(d.d7_retention_base_user), 0),
  IFNULL(SUM(d.d30_retention_user), 0),
  IFNULL(SUM(d.d30_retention_base_user), 0),
  NOW()
FROM country_dashboard_period_metric d
LEFT JOIN (
  SELECT country_code,
    SUM(metric_type = 'COUNTRY_NEW_USER') AS country_new_user,
    SUM(metric_type = 'LUCKY_GIFT_USER') AS lucky_gift_user,
    SUM(metric_type = 'GAME_USER') AS game_user
  FROM country_dashboard_period_user_metric
  WHERE period_type = 'ALL' AND period_key = 'ALL' AND stat_timezone = ? AND sys_origin = ?
  GROUP BY country_code
) u ON u.country_code = d.country_code
WHERE d.period_type = 'DAY' AND d.stat_timezone = ? AND d.sys_origin = ?
GROUP BY d.sys_origin, d.country_code`, statTimezone, statTimezone, sysOrigin, statTimezone, sysOrigin); err != nil {
			return err
		}
	}
	return nil
}

func inPlaceholders(count int) string {
	return "(" + placeholders(count, 1) + ")"
}

func inStringPlaceholders(count int) string {
	return "(" + placeholders(count, 1) + ")"
}

func stringArgs(values []string) []interface{} {
	args := make([]interface{}, 0, len(values))
	for _, value := range values {
		args = append(args, value)
	}
	return args
}