// The error shipping loop — the box-side pump of the Errors tab (spec
// 2026-09-07-errors-and-analytics-design.md §3.2), shaped exactly like
// logshipLoop: the edge and the activator coalesce into one apperr.Sink;
// every tick this drains it and writes one immutable chunk per app under
// apps/<sqid>/<app>/errors/<box>/<hour>/; on each hour boundary it prunes
// its own subtree with the log retention (§3.6: errors follow log
// retention, age and byte cap).
package host

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

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

const errshipTick = 10 * time.Second

func errshipLoop(sink *apperr.Sink, box string) {
	defer func() {
		if r := recover(); r != nil {
			log.Printf("agent: errship loop panic, error shipping stopped for this process: %v", r)
		}
	}()
	ctx := context.Background()
	s, target, err := openStatusStore(ctx, os.Getenv)
	if err != nil {
		if errors.Is(err, errNoStatusStore) {
			log.Printf("agent: errship loop disabled: neither CARLOS_STATUS_BUCKET nor CARLOS_DEPLOYMENT_BUCKET is set")
		} else {
			log.Printf("agent: errship loop disabled: %v", err)
		}
		return
	}
	log.Printf("agent: errship loop armed (10s ticks), publishing to %s", target)

	pending := make(map[applog.AppKey][]apperr.Record)
	pruned := ""
	ticker := time.NewTicker(errshipTick)
	defer ticker.Stop()
	for range ticker.C {
		errshipGather(sink, pending)
		errshipFlush(ctx, s, pending)
		if hour := applog.HourOf(time.Now()); hour != pruned {
			for _, k := range errshipKnownApps(pending, sink) {
				chunkPrune(ctx, s, applog.FamilyErrors, box, k, time.Now())
			}
			pruned = hour
		}
	}
}

// errshipGather drains the sink into pending under the same per-app cap
// logship uses — oldest first.
func errshipGather(sink *apperr.Sink, pending map[applog.AppKey][]apperr.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
	}
}

// errshipFlush writes one create-only chunk per pending app and forgets
// what landed; a failed app keeps its records for the next round.
func errshipFlush(ctx context.Context, s release.Store, pending map[applog.AppKey][]apperr.Record) {
	for k, recs := range pending {
		if len(recs) == 0 {
			delete(pending, k)
			continue
		}
		body, err := apperr.EncodeChunk(recs)
		if err != nil {
			log.Printf("agent: errship encode %s/%s: %v (dropping %d records)", k.Acct, k.App, err, len(recs))
			delete(pending, k)
			continue
		}
		key, err := apperr.ChunkKey(k.Acct, k.App, recs[0].Box, apperr.HourOf(recs[0].TS), recs[0].Seq, len(body))
		if err != nil {
			unshippable(applog.FamilyErrors, k, err)
			delete(pending, k)
			continue
		}
		if _, err := s.Put(ctx, key, body, ""); err != nil {
			log.Printf("agent: errship flush %s/%s: %v (%d records held)", k.Acct, k.App, err, len(recs))
			continue
		}
		delete(pending, k)
	}
}

func errshipKnownApps(pending map[applog.AppKey][]apperr.Record, sink *apperr.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
}