package host

import (
	"context"
	"errors"
	"log"
	"os"
	"time"

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

// usageFlush drains the edge meter and CAS-merges the deltas into the box's
// usage/edge day object, then (when am is non-nil) drains the analytics
// meter's deltas into their app-day objects (spec §3.3: the same tick, a
// second CAS write per app). Usage writes go to s (rebased under
// CARLOS_STATUS_PREFIX on a fleet box); analytics writes go to raw — a day
// file lives under apps/<sqid>/<app>/analytics/, the app's own data, and
// must NOT be rebased (same reasoning as log shipping — logship.go:54-61).
// A failed analytics batch is re-queued for the next tick; a failed usage
// write is returned but does not block the analytics flush.
func usageFlush(ctx context.Context, s, raw release.Store, box string, drain func() map[string]usage.EdgeRow, am *analytics.Meter, now time.Time) error {
	err := usage.UpsertEdgeDay(ctx, s, box, now, drain())
	if am == nil {
		return err
	}
	failed := map[analytics.AppKey][]analytics.Delta{}
	for k, ds := range am.Drain() {
		for _, d := range ds {
			uerr := analytics.UpsertDay(ctx, raw, k.Acct, k.App, d)
			if errors.Is(uerr, applog.ErrBadKeySegment) {
				// No key, so no retry can ever land it: drop, never re-queue.
				unshippable(applog.FamilyAnalytics, k, uerr)
				continue
			}
			if uerr != nil {
				log.Printf("agent: analytics flush %s/%s %s: %v (re-queued)", k.Acct, k.App, d.Date, uerr)
				failed[k] = append(failed[k], d)
			}
		}
	}
	if len(failed) > 0 {
		am.Requeue(failed)
	}
	return err
}

// usageLoop flushes edge usage every 5 minutes, forever — the metricsLoop
// conventions exactly: recover guard, disabled with a log line if the box
// has no status store (mirroring metricsLoop's errNoStatusStore vs. other-error
// distinction at metrics_loop.go:773-779, rather than the brief's single log
// line, so the two loops read the same way in the logs), errors logged and
// never fatal (the drained window is lost, the next one is not; modelling
// grade tolerates the gap). am is the analytics meter (nil disables the
// analytics half of the flush); its deltas are folded in on the same tick.
func usageLoop(drain func() map[string]usage.EdgeRow, am *analytics.Meter) {
	defer func() {
		if r := recover(); r != nil {
			log.Printf("agent: usage loop panic, edge metering stopped for this process: %v", r)
		}
	}()
	ctx := context.Background()
	// Same selection rule statusLoop/metricsLoop use (agent.go's
	// openStatusStore): a status/deployment bucket wins and keeps direct S3;
	// a fleet box with neither set publishes through the edge's store door.
	raw, target, err := openStatusStore(ctx, os.Getenv)
	if err != nil {
		if errors.Is(err, errNoStatusStore) {
			log.Printf("agent: usage loop disabled: neither CARLOS_STATUS_BUCKET nor CARLOS_DEPLOYMENT_BUCKET is set")
		} else {
			log.Printf("agent: usage loop disabled: %v", err)
		}
		return
	}
	s := release.NewPrefixStore(raw, resolveStatusPrefix(os.Getenv))
	box := metricsBoxLabel(os.Getenv, readMachineID())
	log.Printf("agent: usage loop armed (5min flush), publishing to %s", target)
	for {
		time.Sleep(5 * time.Minute)
		if err := usageFlush(ctx, s, raw, box, drain, am, time.Now()); err != nil {
			log.Printf("agent: usage flush: %v", err)
		}
	}
}