package library

import (
	"context"
	"time"
)

// stagePublisher persists a stage tally to scan_runs at a bounded
// cadence: every flushEvery items processed OR every flushPeriod
// elapsed, whichever fires first. The marshal closure produces the
// current tally JSON; persist writes it.
//
// Usage pattern:
//
//	pub := newStagePublisher(ctx, 500, time.Second, marshalFn, persistFn)
//	defer func() {
//	    if err := pub.Flush(); err != nil { /* route to captureErr */ }
//	}()
//	for ... {
//	    processOne()
//	    pub.Tick()
//	}
type stagePublisher struct {
	ctx         context.Context
	flushEvery  int
	flushPeriod time.Duration
	marshal     func() []byte
	persist     func([]byte) error
	sinceItem   int
	lastFlush   time.Time
}

// newStagePublisher constructs a publisher. The ctx is captured for
// future use (currently unused in the helper itself; persist closures
// typically capture their own context via the caller).
func newStagePublisher(ctx context.Context, every int, period time.Duration,
	marshal func() []byte, persist func([]byte) error) *stagePublisher {
	return &stagePublisher{
		ctx:         ctx,
		flushEvery:  every,
		flushPeriod: period,
		marshal:     marshal,
		persist:     persist,
		lastFlush:   time.Now(),
	}
}

// Tick records one item processed. If the flushEvery count is reached
// or flushPeriod has elapsed since the last flush, it triggers a
// persist. Mid-stage flush errors are silently suppressed — partial-
// tally write failures are observability bugs, not scan failures, and
// the next Tick retries.
func (p *stagePublisher) Tick() {
	p.sinceItem++
	if p.sinceItem >= p.flushEvery || time.Since(p.lastFlush) >= p.flushPeriod {
		_ = p.flush()
	}
}

// Flush writes the current tally unconditionally and returns any
// persist error. Call at stage end via a deferred wrapper that
// routes the error to the orchestrator's captureErr — preserves
// today's behavior where final-write failures are diagnostic.
func (p *stagePublisher) Flush() error { return p.flush() }

func (p *stagePublisher) flush() error {
	blob := p.marshal()
	err := p.persist(blob)
	p.sinceItem = 0
	p.lastFlush = time.Now()
	return err
}
