package applog

import (
	"context"
	"strings"
	"sync"
	"time"

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

// readConcurrency bounds one read's in-flight bucket calls: listings in
// the first half, chunk GETs in the second. A day of an app's logs is
// hundreds of chunks, and fetched one at a time that was the whole of the
// Errors tab's ~5 s (2026-09-24, titogo); sixteen at once brings it to
// roughly the slowest few round trips, and bounds what one read can ask of
// the bucket and hold decoded in memory.
const readConcurrency = 16

// ReadChunks walks every box's hour directories under prefix (a
// FamilyPrefix) whose hours intersect [since, until], decodes each chunk as
// T, keeps what passes keep, and returns one slice per box in listing
// order. Hour pruning happens at the PREFIX level — an hour outside the
// window is never listed, which is what makes a 1h read over 7d of history
// cheap (spec §3). collectCap bounds memory per box: past it the OLDEST
// records go and trimmed says so. Store errors fail the read — a partial
// answer that looks whole is the one thing this must never produce.
//
// Listings and GETs run readConcurrency at a time, but the answer is
// assembled strictly in listing order, so it is exactly what a serial walk
// would return. Chunks are fetched at most readConcurrency ahead of the one
// being assembled, so memory stays bounded by collectCap plus that window,
// as it was by collectCap plus one chunk. keep runs on the fetching
// goroutines and must be safe for concurrent use (a Filter's Match is).
func ReadChunks[T any](ctx context.Context, s release.Store, prefix string, since, until time.Time, keep func(T) bool, collectCap int) ([][]T, bool, error) {
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	boxDirs, err := s.ListDirs(ctx, prefix)
	if err != nil {
		return nil, false, err
	}
	hourDirs := make([][]string, len(boxDirs))
	if err := forEachBounded(ctx, len(boxDirs), func(i int) error {
		dirs, err := s.ListDirs(ctx, boxDirs[i])
		if err != nil {
			return err
		}
		for _, d := range dirs { // lexically sorted = chronological
			if hourIntersects(HourNameOfDir(d), since, until) {
				hourDirs[i] = append(hourDirs[i], d)
			}
		}
		return nil
	}); err != nil {
		return nil, false, err
	}
	type hour struct {
		box int
		dir string
	}
	var hours []hour
	for b, dirs := range hourDirs {
		for _, d := range dirs {
			hours = append(hours, hour{b, d})
		}
	}
	hourKeys := make([][]string, len(hours))
	if err := forEachBounded(ctx, len(hours), func(i int) error {
		keys, err := s.List(ctx, hours[i].dir)
		hourKeys[i] = keys // lexically sorted = seq order
		return err
	}); err != nil {
		return nil, false, err
	}
	type chunk struct {
		box int
		key string
	}
	var chunks []chunk
	for i, keys := range hourKeys {
		for _, k := range keys {
			chunks = append(chunks, chunk{hours[i].box, k})
		}
	}

	// Fetch ahead, assemble in order. A slot's channel is buffered, so a
	// fetch whose result is never read (the read failed first) still ends;
	// the feeder stops at cancel.
	type fetched struct {
		recs []T
		err  error
	}
	slots := make([]chan fetched, len(chunks))
	for i := range slots {
		slots[i] = make(chan fetched, 1)
	}
	tokens := make(chan struct{}, readConcurrency)
	go func() {
		for i, c := range chunks {
			select {
			case tokens <- struct{}{}:
			case <-ctx.Done():
				return
			}
			go func() {
				body, _, err := s.Get(ctx, c.key)
				if err != nil {
					slots[i] <- fetched{err: err}
					return
				}
				all, err := DecodeJSONL[T](body)
				if err != nil {
					slots[i] <- fetched{err: err}
					return
				}
				var kept []T
				for _, r := range all {
					if keep(r) {
						kept = append(kept, r)
					}
				}
				slots[i] <- fetched{recs: kept}
			}()
		}
	}()
	trimmed := false
	perBox := make([][]T, len(boxDirs))
	for i, c := range chunks {
		var f fetched
		select {
		case f = <-slots[i]:
		case <-ctx.Done():
			return nil, false, ctx.Err()
		}
		<-tokens
		if f.err != nil {
			return nil, false, f.err
		}
		recs := append(perBox[c.box], f.recs...)
		if over := len(recs) - collectCap; over > 0 {
			recs = recs[over:]
			trimmed = true
		}
		perBox[c.box] = recs
	}
	return perBox, trimmed, nil
}

// forEachBounded runs fn(0..n-1) readConcurrency at a time and returns the
// first error, cancelling nothing itself: fn's own ctx is the caller's, and
// ReadChunks cancels it on return.
func forEachBounded(ctx context.Context, n int, fn func(i int) error) error {
	var (
		wg       sync.WaitGroup
		mu       sync.Mutex
		firstErr error
	)
	tokens := make(chan struct{}, readConcurrency)
	for i := 0; i < n; i++ {
		mu.Lock()
		failed := firstErr != nil
		mu.Unlock()
		if failed {
			break // the read has already failed; start nothing more
		}
		select {
		case tokens <- struct{}{}:
		case <-ctx.Done():
			wg.Wait()
			return ctx.Err()
		}
		wg.Add(1)
		go func() {
			defer func() { <-tokens; wg.Done() }()
			if err := fn(i); err != nil {
				mu.Lock()
				if firstErr == nil {
					firstErr = err
				}
				mu.Unlock()
			}
		}()
	}
	wg.Wait()
	return firstErr
}

// ReadWindow assembles one app's merged timeline from the bucket: every
// box's chunks whose hour directories intersect [f.Since, f.Until], decoded,
// filtered through f, merged in (ts, box, seq) order. Hour pruning happens
// at the PREFIX level — an hour outside the window is never listed, which
// is what makes a 1h read over 7d of history cheap (spec §3).
//
// collectCap bounds memory against a pathological window: past it the
// OLDEST records go (the tail is what a log reader came for), and the
// second return says so. Store errors fail the read — a partial answer
// that looks whole is the one thing this must never produce.
func ReadWindow(ctx context.Context, s release.Store, acct, app string, f Filter, collectCap int) ([]Record, bool, error) {
	prefix, err := Prefix(acct, app)
	if err != nil {
		return nil, false, err
	}
	perBox, trimmed, err := ReadChunks[Record](ctx, s, prefix, f.Since, f.Until, f.Match, collectCap)
	if err != nil {
		return nil, false, err
	}
	merged := MergeRecords(perBox...)
	if over := len(merged) - collectCap; over > 0 {
		merged = merged[over:]
		trimmed = true
	}
	return merged, trimmed, nil
}

// hourIntersects reports whether the hour directory named hour overlaps
// [since, until] (zero = unbounded). Unparsable names never match — an
// alien directory is not silently read as data.
func hourIntersects(hour string, since, until time.Time) bool {
	start, err := time.ParseInLocation(hourFormat, hour, time.UTC)
	if err != nil {
		return false
	}
	end := start.Add(time.Hour)
	if !since.IsZero() && !end.After(since) {
		return false
	}
	if !until.IsZero() && start.After(until) {
		return false
	}
	return true
}

// HourNameOfDir extracts the final path segment of a "/"-terminated prefix.
func HourNameOfDir(dir string) string {
	trimmed := strings.TrimSuffix(dir, "/")
	if i := strings.LastIndexByte(trimmed, '/'); i >= 0 {
		return trimmed[i+1:]
	}
	return trimmed
}