// Package batch accumulates telemetry readings and flushes them in bulk,
// either once a size threshold is hit or on a fixed interval — so ClickHouse
// gets bulk inserts instead of one round-trip per reading.
package batch

import (
	"context"
	"log/slog"
	"sync"
	"time"

	"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/telemetry"
)

// FlushFunc persists one batch. It is called with a context bounded by
// FlushTimeout, both for interval-triggered and shutdown-triggered flushes.
type FlushFunc func(ctx context.Context, readings []telemetry.Reading) error

// Batcher buffers readings in memory and flushes them via FlushFunc when
// MaxSize is reached or Interval elapses, whichever comes first.
type Batcher struct {
	maxSize      int
	interval     time.Duration
	flushTimeout time.Duration
	flush        FlushFunc
	logger       *slog.Logger

	mu  sync.Mutex
	buf []telemetry.Reading

	stop chan struct{}
	done chan struct{}
}

func New(maxSize int, interval, flushTimeout time.Duration, flush FlushFunc, logger *slog.Logger) *Batcher {
	return &Batcher{
		maxSize:      maxSize,
		interval:     interval,
		flushTimeout: flushTimeout,
		flush:        flush,
		logger:       logger,
	}
}

// Start begins the interval-flush loop. Call once before the first Add.
func (b *Batcher) Start() {
	b.stop = make(chan struct{})
	b.done = make(chan struct{})
	go b.loop()
}

func (b *Batcher) loop() {
	defer close(b.done)
	ticker := time.NewTicker(b.interval)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			b.flushNow("interval")
		case <-b.stop:
			b.flushNow("shutdown")
			return
		}
	}
}

// Add appends a reading, flushing immediately if the buffer just reached
// maxSize rather than waiting for the next tick.
func (b *Batcher) Add(r telemetry.Reading) {
	b.mu.Lock()
	b.buf = append(b.buf, r)
	full := len(b.buf) >= b.maxSize
	b.mu.Unlock()

	if full {
		b.flushNow("size")
	}
}

// Stop flushes any remaining buffered readings and waits for the flush loop
// to exit. Safe to call once, after Start.
func (b *Batcher) Stop() {
	close(b.stop)
	<-b.done
}

func (b *Batcher) flushNow(reason string) {
	b.mu.Lock()
	if len(b.buf) == 0 {
		b.mu.Unlock()
		return
	}
	readings := b.buf
	b.buf = nil
	b.mu.Unlock()

	ctx, cancel := context.WithTimeout(context.Background(), b.flushTimeout)
	defer cancel()

	if err := b.flush(ctx, readings); err != nil {
		b.logger.Error("batch flush failed", "reason", reason, "size", len(readings), "error", err)
		return
	}
	b.logger.Debug("batch flushed", "reason", reason, "size", len(readings))
}