package analytics

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"strings"

	"github.com/carlosframework/platform/internal/applog"
	"github.com/carlosframework/platform/internal/release"
)

// HourRow is one UTC hour's counters.
type HourRow struct {
	PV    int64 `json:"pv"`
	Bots  int64 `json:"bots"`
	Bytes int64 `json:"bytes"`
}

// DropCounts is why events were dropped at a door (steps 3–4 fill them).
type DropCounts struct {
	Rate      int64 `json:"rate"`
	Size      int64 `json:"size"`
	Origin    int64 `json:"origin"`
	Malformed int64 `json:"malformed"`
}

// Dropped is the honesty block (spec §3.3).
type Dropped struct {
	E     DropCounts `json:"e"`
	A     DropCounts `json:"a"`
	Flush int64      `json:"flush"`
}

// Day is apps/<sqid>/<app>/analytics/<box>/days/<date>.json (spec §3.3):
// one box's counting for one app for one UTC day. Additive shape.
type Day struct {
	V         int                `json:"v"`
	Date      string             `json:"date"`
	Hours     map[string]HourRow `json:"hours"`
	Visitors  string             `json:"visitors"`
	Pages     map[string]int64   `json:"pages"`
	Referers  map[string]int64   `json:"referers"`
	Events    map[string]int64   `json:"events"`
	Seq       map[string]int64   `json:"seq"`
	Dropped   Dropped            `json:"dropped"`
	Countries map[string]int64   `json:"countries"`
	// GeoSources counts pageviews per database type that answered (geoip
	// spec §9): the provenance the attribution on the analytics page needs.
	GeoSources map[string]int64 `json:"geo_sources,omitempty"`
	Restarted  bool             `json:"restarted,omitempty"`
}

// Delta is one flush batch from one box for one app-day.
type Delta struct {
	Date      string
	Box       string
	Seq       int64
	Hours     map[string]HourRow
	Visitors  *HLL
	Pages     *TopK
	Referers  *TopK
	Events    *TopK
	Dropped   Dropped
	Restarted bool
	// Countries and GeoSources are pageview counts by GeoIP answer (geoip
	// spec §9), bounded by boundedAdd.
	Countries  map[string]int64
	GeoSources map[string]int64

	// requeued marks a delta that has already been through one failed
	// flush (internal/analytics.Meter.Requeue). Not serialised: it never
	// reaches the day file, only decides whether a second Requeue keeps
	// or drops the delta.
	requeued bool
}

// Prefix is the app's analytics/ subtree. An invalid account or app is
// refused (applog.ErrBadKeySegment), never built.
func Prefix(sqid, app string) (string, error) {
	return applog.FamilyPrefix(applog.FamilyAnalytics, sqid, app)
}

// BoxPrefix is one box's day files.
func BoxPrefix(sqid, app, box string) (string, error) {
	p, err := applog.FamilyBoxPrefix(applog.FamilyAnalytics, sqid, app, box)
	if err != nil {
		return "", err
	}
	return p + "days/", nil
}

// DayKey is one day file.
func DayKey(sqid, app, box, date string) (string, error) {
	p, err := BoxPrefix(sqid, app, box)
	if err != nil {
		return "", err
	}
	if err := applog.CheckSegment("date", date); err != nil {
		return "", err
	}
	return p + date + ".json", nil
}

// Merge folds x into d unless x.Seq is not above the recorded seq for
// x.Box — the idempotency rule that makes a retried conditional put safe.
func (d *Day) Merge(x Delta) (bool, error) {
	if x.Seq <= d.Seq[x.Box] {
		return false, nil
	}
	if d.Seq == nil {
		d.Seq = map[string]int64{}
	}
	d.V, d.Date = 1, x.Date
	if d.Hours == nil {
		d.Hours = map[string]HourRow{}
	}
	for h, row := range x.Hours {
		cur := d.Hours[h]
		cur.PV += row.PV
		cur.Bots += row.Bots
		cur.Bytes += row.Bytes
		d.Hours[h] = cur
	}
	// A corrupt or unreadable sketch must not poison the whole day: start
	// fresh rather than fail the merge. The bad string is overwritten by
	// the freshly encoded sketch below either way.
	vis, err := DecodeHLL(d.Visitors)
	if err != nil {
		vis = &HLL{}
	}
	if x.Visitors != nil {
		vis.Merge(x.Visitors)
	}
	d.Visitors = vis.Encode()
	d.Pages = mergeTop(d.Pages, x.Pages)
	d.Referers = mergeTop(d.Referers, x.Referers)
	d.Events = mergeTop(d.Events, x.Events)
	d.Dropped.E = addDrops(d.Dropped.E, x.Dropped.E)
	d.Dropped.A = addDrops(d.Dropped.A, x.Dropped.A)
	d.Dropped.Flush += x.Dropped.Flush
	d.Restarted = d.Restarted || x.Restarted
	d.Countries = mergeCounts(d.Countries, x.Countries, maxCountryKeys, nil)
	d.GeoSources = mergeCounts(d.GeoSources, x.GeoSources, maxGeoSourceKeys, attributed)
	d.Seq[x.Box] = x.Seq
	return true, nil
}

func mergeTop(m map[string]int64, t *TopK) map[string]int64 {
	k := TopKFrom(m)
	if t != nil {
		k.Merge(t)
	}
	return k.Fold()
}

func addDrops(a, b DropCounts) DropCounts {
	return DropCounts{a.Rate + b.Rate, a.Size + b.Size, a.Origin + b.Origin, a.Malformed + b.Malformed}
}

// UpsertDay CAS-merges x into the app-day object for x.Box: read (etag),
// Merge, conditional put; on ErrCASConflict re-read and retry, three
// attempts. A no-op merge (seq already recorded) writes nothing.
func UpsertDay(ctx context.Context, s release.Store, sqid, app string, x Delta) error {
	key, err := DayKey(sqid, app, x.Box, x.Date)
	if err != nil {
		return err
	}
	for attempt := 0; attempt < 3; attempt++ {
		var day Day
		raw, etag, err := s.Get(ctx, key)
		switch {
		case err == nil:
			if uerr := json.Unmarshal(raw, &day); uerr != nil {
				return fmt.Errorf("analytics: parse %s: %w", key, uerr)
			}
		case errors.Is(err, os.ErrNotExist):
			etag = ""
		default:
			return err
		}
		applied, err := day.Merge(x)
		if err != nil {
			return err
		}
		if !applied {
			return nil
		}
		out, err := json.Marshal(day)
		if err != nil {
			return err
		}
		if _, err := s.Put(ctx, key, out, etag); err == nil {
			return nil
		} else if !errors.Is(err, release.ErrCASConflict) {
			return err
		}
	}
	return fmt.Errorf("analytics: %s: lost CAS race three times", key)
}

const (
	maxCountryKeys   = 300
	maxGeoSourceKeys = 8
)

// attributed reports whether a database type carries an attribution the
// analytics page must keep showing (geoip spec §9, §13: the DB-IP and
// GeoLite2 prefixes). Such a source is never folded into _other, so its
// credit survives any number of other types (codex R7-3). Kept here as two
// prefixes rather than importing internal/geoip, which this package's
// dependants should not need.
func attributed(source string) bool {
	return strings.HasPrefix(source, "DBIP-") || strings.HasPrefix(source, "GeoLite2-")
}

// boundedAdd adds n to m[k], folding a new key past max into OtherKey
// unless keep says it must never fold.
func boundedAdd(m map[string]int64, k string, n int64, max int, keep func(string) bool) {
	if _, ok := m[k]; ok || k == OtherKey || len(m) < max || (keep != nil && keep(k)) {
		m[k] += n
		return
	}
	m[OtherKey] += n
}

func mergeCounts(dst, src map[string]int64, max int, keep func(string) bool) map[string]int64 {
	if dst == nil {
		dst = map[string]int64{}
	}
	for k, n := range src {
		boundedAdd(dst, k, n, max, keep)
	}
	return dst
}

// countGeo applies spec §9 to one pageview: hit counts its country (or
// _unknown for a coordinate-only answer), miss counts _unknown, off and ""
// count nothing; a hit or miss also counts its source.
func countGeo(d *Delta, o Obs) {
	if o.GeoState != "hit" && o.GeoState != "miss" {
		return
	}
	c := o.Country
	if o.GeoState == "miss" || c == "" {
		c = "_unknown"
	}
	boundedAdd(d.Countries, c, 1, maxCountryKeys, nil)
	if o.GeoSource != "" {
		boundedAdd(d.GeoSources, o.GeoSource, 1, maxGeoSourceKeys, attributed)
	}
}