// The log shipping loop — the box-side pump of `carlos logs` (spec
// 2026-08-10-carlos-logs-design.md §5). Producers (the activator's exec
// tees and platform events) buffer into one applog.Sink; every 10s — or
// within ~2s of a platform event, which is what lets `carlos deploy`'s
// timeout window see the wake error — this loop drains it and writes one
// immutable gzipped chunk per app into the deployment bucket. On each hour
// boundary it prunes ITS OWN box subtree per app: hours beyond the age
// floor, then oldest-first beyond the byte cap (sizes ride in the chunk
// keys — applog.ChunkKeySize — so no Store interface change). The
// sampler/flush split mirrors metricsLoop for the same reason: the loop
// shares a process with the box's traffic, so it recovers panics and
// disables itself without a bucket rather than taking the box down.
package host

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

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

const (
	logshipTick     = 10 * time.Second
	logshipDebounce = 2 * time.Second // event-poke settle: batch a wake's burst into one PUT
	logshipMaxAge   = 7 * 24 * time.Hour
	logshipMaxBytes = 50 << 20 // per app per box
	// logshipMaxPending caps records held across failed flushes, per app.
	// The bucket being unwritable must not grow the agent's heap forever
	// (samplerMaxBuf's reasoning); oldest go first, silently — a box-wide
	// bucket outage is loudly visible elsewhere, and the gap in seq numbers
	// records that something is missing.
	logshipMaxPending = 4096
)

func logshipLoop(sink *applog.Sink, box string) {
	// Same rule as metricsLoop: this loop shares a process with the agent
	// serving traffic — log shipping stopping is a bad day, the box
	// stopping is an outage.
	defer func() {
		if r := recover(); r != nil {
			log.Printf("agent: logship loop panic, log shipping stopped for this process: %v", r)
		}
	}()

	ctx := context.Background()
	// Same store selection as statusLoop/metricsLoop (agent.go's
	// openStatusStore): a bucket wins, and a cut-over fleet box with no bucket
	// ships through the edge's store door instead of going silent.
	//
	// The keys stay UNPREFIXED — deliberately unlike status and metrics, which
	// rebase under CARLOS_STATUS_PREFIX. A log chunk is the app's own data
	// (applog.Prefix = release.AppPrefix + "logs/"), and the console reads it
	// straight from the platform layout (applog.ReadWindow over
	// apps/<sqid>/<app>/logs/**); rebasing under fleets/<name>/ would file a
	// fleet box's logs somewhere `carlos logs` never looks. The door's write
	// ACL admits exactly this shape for the box's own account (aclWrite's
	// segment-exact apps/<sqid>/<app>/logs/** scope).
	s, target, err := openStatusStore(ctx, os.Getenv)
	if err != nil {
		if errors.Is(err, errNoStatusStore) {
			log.Printf("agent: logship loop disabled: neither CARLOS_STATUS_BUCKET nor CARLOS_DEPLOYMENT_BUCKET is set")
		} else {
			log.Printf("agent: logship loop disabled: %v", err)
		}
		return
	}
	log.Printf("agent: logship loop armed (10s ticks, event fast-flush), publishing to %s", target)

	pending := make(map[applog.AppKey][]applog.Record)
	pruned := "" // last hour a prune ran for, "" = not yet
	ticker := time.NewTicker(logshipTick)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
		case <-sink.Wake():
			time.Sleep(logshipDebounce) // let the event's burst finish
		}
		logshipGather(sink, pending)
		logshipFlush(ctx, s, pending)

		if hour := applog.HourOf(time.Now()); hour != pruned {
			// Prune covers every app this box is CURRENTLY holding or has
			// just flushed; an app that stops logging entirely ages out via
			// other boxes' passes or the app's own deletion sweep. Sub-cap
			// apps pay one ListDirs each per hour.
			for _, k := range logshipKnownApps(pending, sink) {
				chunkPrune(ctx, s, applog.FamilyLogs, box, k, time.Now())
			}
			pruned = hour
		}
	}
}

// logshipGather drains the sink into pending, enforcing the per-app cap —
// oldest first, so what survives a long bucket outage is the freshest tail.
func logshipGather(sink *applog.Sink, pending map[applog.AppKey][]applog.Record) {
	for k, recs := range sink.Drain() {
		merged := append(pending[k], recs...)
		if over := len(merged) - logshipMaxPending; over > 0 {
			merged = merged[over:]
		}
		pending[k] = merged
	}
}

// logshipFlush writes one chunk per pending app — create-only, keyed by the
// batch's first seq, so a retried batch can never overwrite a delivered one
// — and forgets what landed. A failed app keeps its records for the next
// round; other apps' flushes proceed regardless.
func logshipFlush(ctx context.Context, s release.Store, pending map[applog.AppKey][]applog.Record) {
	for k, recs := range pending {
		if len(recs) == 0 {
			delete(pending, k)
			continue
		}
		body, err := applog.EncodeChunk(recs)
		if err != nil { // marshal of our own records — a bug, not a retry case
			log.Printf("agent: logship encode %s/%s: %v (dropping %d records)", k.Acct, k.App, err, len(recs))
			delete(pending, k)
			continue
		}
		key, err := applog.ChunkKey(k.Acct, k.App, recs[0].Box, applog.HourOf(recs[0].TS), recs[0].Seq, len(body))
		if err != nil {
			unshippable(applog.FamilyLogs, k, err)
			delete(pending, k)
			continue
		}
		if _, err := s.Put(ctx, key, body, ""); err != nil {
			log.Printf("agent: logship flush %s/%s: %v (%d records held)", k.Acct, k.App, err, len(recs))
			continue
		}
		delete(pending, k)
	}
}

// logshipKnownApps is the union of apps with pending records and apps the
// sink has ever buffered for — the prune worklist.
func logshipKnownApps(pending map[applog.AppKey][]applog.Record, sink *applog.Sink) []applog.AppKey {
	seen := make(map[applog.AppKey]bool)
	var out []applog.AppKey
	for k := range pending {
		if !seen[k] {
			seen[k] = true
			out = append(out, k)
		}
	}
	for _, k := range sink.Known() {
		if !seen[k] {
			seen[k] = true
			out = append(out, k)
		}
	}
	return out
}

// chunkPrune enforces retention for ONE app's ONE family on THIS box's
// subtree only — each box owns what it wrote (spec §5). Hour granularity: an
// hour goes when it is older than the age floor, or (oldest first) while the
// subtree's key-embedded sizes total over the byte cap. Errors are logged
// and left for the next hourly pass — retention is idempotent.
func chunkPrune(ctx context.Context, s release.Store, fam applog.Family, box string, k applog.AppKey, now time.Time) {
	prefix, err := applog.FamilyBoxPrefix(fam, k.Acct, k.App, box)
	if err != nil {
		unshippable(fam, k, err)
		return
	}
	hourDirs, err := s.ListDirs(ctx, prefix)
	if err != nil {
		log.Printf("agent: logship prune %s: %v", prefix, err)
		return
	}
	// ListDirs returns full prefixes ending in "/", sorted is not promised —
	// the hour names sort chronologically, so order them ourselves via the
	// keys' lexical shape. Collect per-hour keys+sizes as we go.
	type hourEnt struct {
		dir   string
		keys  []string
		bytes int64
	}
	var hours []hourEnt
	var total int64
	for _, dir := range hourDirs {
		keys, err := s.List(ctx, dir)
		if err != nil {
			log.Printf("agent: logship prune %s: %v", dir, err)
			return
		}
		var b int64
		for _, key := range keys {
			if n, ok := applog.ChunkKeySize(key); ok {
				b += n
			}
		}
		hours = append(hours, hourEnt{dir: dir, keys: keys, bytes: b})
		total += b
	}
	// hourDirs from both implementations come lexically sorted (S3 lists in
	// key order; DirStore sorts), and hour names are zero-padded UTC — so
	// hours[0] is oldest.
	cutoff := now.Add(-logshipMaxAge)
	for _, h := range hours {
		hourName := applog.HourNameOfDir(h.dir)
		drop := applog.HourBefore(hourName, cutoff) || total > logshipMaxBytes
		if !drop {
			break // hours are oldest-first; nothing further is droppable
		}
		for _, key := range h.keys {
			if err := s.Delete(ctx, key); err != nil {
				log.Printf("agent: logship prune %s: %v", key, err)
				return
			}
		}
		total -= h.bytes
	}
}

// unshippableSeen remembers which (family, app) pairs unshippable has
// already reported, so a route that can never ship says so once per
// process rather than every tick. Bounded by the routes a box serves.
var unshippableSeen sync.Map

type unshippableKey struct {
	fam applog.Family
	app applog.AppKey
}

// unshippable reports, once per family and app, that the family's key
// builder refused the app (applog.ErrBadKeySegment): the records are
// dropped, and no write or prune happens for it. The case that put this
// here is a route with no app, filed under its account with an empty app
// name; writing it would make apps/<sqid>//..., which hostops.Apps refuses
// to walk.
func unshippable(fam applog.Family, k applog.AppKey, err error) {
	if _, dup := unshippableSeen.LoadOrStore(unshippableKey{fam, k}, true); dup {
		return
	}
	log.Printf("agent: %s: not shipping for account %q app %q, dropping its records (logged once): %v", fam, k.Acct, k.App, err)
}