package applog

import (
	"bufio"
	"bytes"
	"compress/gzip"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"strconv"
	"strings"
	"time"

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

// hourFormat is the UTC hour vocabulary in chunk keys — deliberately the
// same format internal/metrics uses for its raw chunks, so the bucket
// speaks one hour dialect.
const hourFormat = "2006-01-02-15"

// HourOf formats t as the UTC hour directory a chunk belongs to.
func HourOf(t time.Time) string { return t.UTC().Format(hourFormat) }

// HourBefore reports whether hour (a HourOf-formatted directory name) ends
// strictly before cutoff's own hour — the retention test. An unparsable
// name never matches: pruning must not delete on a guess.
func HourBefore(hour string, cutoff time.Time) bool {
	t, err := time.ParseInLocation(hourFormat, hour, time.UTC)
	if err != nil {
		return false
	}
	return t.Add(time.Hour).Sub(cutoff.UTC()) <= 0 && HourOf(cutoff) != hour
}

// Family names one chunk family under an app — the path segment after
// apps/<sqid>/<app>/. logs/ is the original; errors/ and analytics/ are
// the siblings the errors-and-analytics spec adds (§3.2). The key shape
// below the family segment is identical for every family, which is what
// lets one prune loop, one store-door grant and one reader serve all three.
type Family string

const (
	FamilyLogs      Family = "logs"
	FamilyErrors    Family = "errors"
	FamilyAnalytics Family = "analytics"
)

// ErrBadKeySegment is every family builder's refusal: an account or app
// that is not a valid name, or a box or hour that is empty or carries a
// slash. Writers skip the write on it; readers fail the read. A bad
// segment never becomes a key, so no box can write apps/<sqid>//... (an
// app-less route once did exactly that, and hostops.Apps, rightly, refuses
// to walk a bucket holding it).
var ErrBadKeySegment = errors.New("applog: refusing to build a key from an empty or invalid segment")

// CheckApp is the account and app half of every family builder's check:
// the same names hostops.Apps accepts when it walks apps/.
func CheckApp(acct, app string) error {
	if !release.ValidAccount(acct) || !release.ValidApp(app) {
		return fmt.Errorf("%w: account %q, app %q", ErrBadKeySegment, acct, app)
	}
	return nil
}

// CheckSegment refuses a box, hour or date segment that would collapse or
// split a path level. internal/analytics uses it for its day-file leaf.
func CheckSegment(what, v string) error {
	if v == "" || strings.ContainsRune(v, '/') {
		return fmt.Errorf("%w: %s %q", ErrBadKeySegment, what, v)
	}
	return nil
}

// FamilyPrefix is the app's whole subtree for one family — under
// release.AppPrefix on purpose (spec §3): it is the app's data, so the
// owner gate maps onto the prefix and app trash/purge sweeps it for free.
// An invalid account or app is refused (ErrBadKeySegment), never built.
func FamilyPrefix(fam Family, acct, app string) (string, error) {
	if err := CheckApp(acct, app); err != nil {
		return "", err
	}
	return release.AppPrefix(acct, app) + string(fam) + "/", nil
}

// FamilyBoxPrefix is one box's subtree within a family — the unit of
// write-side ownership: each box prunes only what it wrote.
func FamilyBoxPrefix(fam Family, acct, app, box string) (string, error) {
	p, err := FamilyPrefix(fam, acct, app)
	if err != nil {
		return "", err
	}
	if err := CheckSegment("box", box); err != nil {
		return "", err
	}
	return p + box + "/", nil
}

// FamilyHourPrefix is one UTC hour within a box's subtree.
func FamilyHourPrefix(fam Family, acct, app, box, hour string) (string, error) {
	p, err := FamilyBoxPrefix(fam, acct, app, box)
	if err != nil {
		return "", err
	}
	if err := CheckSegment("hour", hour); err != nil {
		return "", err
	}
	return p + hour + "/", nil
}

// FamilyChunkKey names one flushed chunk in any family. See ChunkKey for
// why seq and size live in the key.
func FamilyChunkKey(fam Family, acct, app, box, hour string, seq uint64, size int) (string, error) {
	p, err := FamilyHourPrefix(fam, acct, app, box, hour)
	if err != nil {
		return "", err
	}
	return fmt.Sprintf("%s%016d-%d.jsonl.gz", p, seq, size), nil
}

// Prefix is the app's whole log subtree: FamilyPrefix for logs.
func Prefix(acct, app string) (string, error) { return FamilyPrefix(FamilyLogs, acct, app) }

// BoxPrefix is one box's subtree within an app's logs.
func BoxPrefix(acct, app, box string) (string, error) {
	return FamilyBoxPrefix(FamilyLogs, acct, app, box)
}

// HourPrefix is one UTC hour within a box's log subtree.
func HourPrefix(acct, app, box, hour string) (string, error) {
	return FamilyHourPrefix(FamilyLogs, acct, app, box, hour)
}

// ChunkKey names one flushed log chunk: seq is the first record's per-box
// sequence number, zero-padded (16 digits — seq is milliseconds-based, see
// NewSink) so lexical listing order is numeric order, and size is the
// object's own byte length. Size lives IN the key on purpose:
// release.Store's List returns keys only, and retention's byte cap needs
// sizes — embedding them costs a suffix, extending the Store interface
// costs every implementation. Chunks are written once and never appended
// or rewritten (spec §3).
func ChunkKey(acct, app, box, hour string, seq uint64, size int) (string, error) {
	return FamilyChunkKey(FamilyLogs, acct, app, box, hour, seq, size)
}

// ChunkKeySize recovers the byte size a ChunkKey carries, false for any key
// not shaped like one (retention treats unparsable as size-unknown = 0 —
// age still prunes it, the byte cap just cannot count it).
func ChunkKeySize(key string) (int64, bool) {
	base := key
	if i := strings.LastIndexByte(base, '/'); i >= 0 {
		base = base[i+1:]
	}
	base, ok := strings.CutSuffix(base, ".jsonl.gz")
	if !ok {
		return 0, false
	}
	_, sizeStr, ok := strings.Cut(base, "-")
	if !ok {
		return 0, false
	}
	n, err := strconv.ParseInt(sizeStr, 10, 64)
	if err != nil || n < 0 {
		return 0, false
	}
	return n, true
}

// EncodeJSONL renders any record shape as gzipped JSONL, one per line —
// the chunk codec every family shares.
func EncodeJSONL[T any](recs []T) ([]byte, error) {
	var buf bytes.Buffer
	zw := gzip.NewWriter(&buf)
	enc := json.NewEncoder(zw)
	for _, r := range recs {
		if err := enc.Encode(r); err != nil {
			return nil, err
		}
	}
	if err := zw.Close(); err != nil {
		return nil, err
	}
	return buf.Bytes(), nil
}

// DecodeJSONL parses a chunk, skipping malformed lines rather than failing
// the whole object — one corrupt line must not cost the member the rest of
// the hour. Corrupt gzip is an error: that is a broken object, not a line.
func DecodeJSONL[T any](b []byte) ([]T, error) {
	zr, err := gzip.NewReader(bytes.NewReader(b))
	if err != nil {
		return nil, err
	}
	defer zr.Close()
	var out []T
	sc := bufio.NewScanner(zr)
	sc.Buffer(nil, 1<<20) // lines are capped well below this at the tee
	for sc.Scan() {
		var r T
		if json.Unmarshal(sc.Bytes(), &r) != nil {
			continue
		}
		out = append(out, r)
	}
	if err := sc.Err(); err != nil && err != io.EOF {
		return out, err
	}
	return out, nil
}

// EncodeChunk renders log records: EncodeJSONL over Record.
func EncodeChunk(recs []Record) ([]byte, error) { return EncodeJSONL(recs) }

// DecodeChunk parses a log chunk: DecodeJSONL over Record.
func DecodeChunk(b []byte) ([]Record, error) { return DecodeJSONL[Record](b) }