package dbreads

import (
	"bytes"
	"context"
	"crypto/ecdh"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"hash"
	"sync"
	"time"

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

var errCapped = errors.New("dbreads: the output cap was reached")

// putCreateOnly creates key with body, retrying until it lands or ctx ends.
// An error or a conflict is resolved by reading the key back: the same
// bytes mean an earlier attempt landed (spec §5.4). The caller keeps body
// until this returns, so a retry is the same bytes.
func putCreateOnly(ctx context.Context, s release.Store, key string, body []byte) error {
	wait := 250 * time.Millisecond
	for {
		_, err := s.Put(ctx, key, body, "")
		if err == nil {
			return nil
		}
		got, _, gerr := s.Get(ctx, key)
		switch {
		case gerr == nil && bytes.Equal(got, body):
			return nil
		case gerr == nil:
			return fmt.Errorf("dbreads: %s holds other bytes: %w", key, release.ErrCASConflict)
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(wait):
		}
		wait = min(wait*2, 5*time.Second)
	}
}

// chunkWriter seals a read's output into out/<seq> objects: one every
// size bytes (ChunkBytes for a query, PullChunkBytes for a pull), and whatever is pending every `every`; close seals what is
// left, possibly nothing, as the last chunk. Past max nothing more is kept
// and Write answers errCapped. Each put is create-only and confirmed before
// the next is sealed, under the lock, so seqs are in order (the run
// chunker's discipline).
type chunkWriter struct {
	ctx    context.Context
	store  release.Store
	id     string
	pub    *ecdh.PublicKey
	stream string
	max    int64
	size   int

	mu       sync.Mutex
	pending  []byte
	seq      int64
	total    int64
	sum      hash.Hash
	err      error
	isCapped bool

	stop chan struct{}
	done chan struct{}
}

func newChunkWriter(ctx context.Context, s release.Store, id string, pub *ecdh.PublicKey, stream string, every time.Duration, max int64, size int) *chunkWriter {
	c := &chunkWriter{ctx: ctx, store: s, id: id, pub: pub, stream: stream, max: max, size: size, sum: sha256.New(),
		stop: make(chan struct{}), done: make(chan struct{})}
	go c.tick(every)
	return c
}

func (c *chunkWriter) tick(every time.Duration) {
	defer close(c.done)
	t := time.NewTicker(every)
	defer t.Stop()
	for {
		select {
		case <-c.stop:
			return
		case <-t.C:
			c.mu.Lock()
			if len(c.pending) > 0 && c.err == nil {
				c.err = c.flushLocked(c.pending, false)
				c.pending = nil
			}
			c.mu.Unlock()
		}
	}
}

func (c *chunkWriter) Write(p []byte) (int, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.err != nil {
		return 0, c.err
	}
	if c.isCapped {
		return 0, errCapped
	}
	capped := false
	if room := c.max - c.total; int64(len(p)) > room {
		p, capped = p[:room], true
	}
	c.sum.Write(p)
	c.total += int64(len(p))
	c.pending = append(c.pending, p...)
	for len(c.pending) >= c.size {
		if err := c.flushLocked(c.pending[:c.size], false); err != nil {
			c.err = err
			return 0, err
		}
		c.pending = append([]byte(nil), c.pending[c.size:]...)
	}
	if capped {
		c.isCapped = true
		return len(p), errCapped
	}
	return len(p), nil
}

// flushLocked seals and creates the next chunk. seq moves only once the
// create is confirmed, so close never reports, and no result ever names, a
// chunk that was not confirmed (spec §5.4; plan review finding 9). A chunk
// whose create stayed ambiguous until the context ended may land later at
// seq+1; no result names it, so no reader relays it.
func (c *chunkWriter) flushLocked(b []byte, last bool) error {
	next := c.seq + 1
	sealed, err := SealChunk(c.pub, c.id, next, c.stream, last, b)
	if err != nil {
		return err
	}
	if err := putCreateOnly(c.ctx, c.store, ChunkKey(c.id, next), sealed); err != nil {
		return err
	}
	c.seq = next
	return nil
}

func (c *chunkWriter) capped() bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.isCapped
}

// close seals the pending bytes as the last chunk and reports the last seq,
// the plaintext byte count and its SHA-256.
func (c *chunkWriter) close() (int64, int64, string, error) {
	close(c.stop)
	<-c.done
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.err == nil {
		c.err = c.flushLocked(c.pending, true)
		c.pending = nil
	}
	return c.seq, c.total, hex.EncodeToString(c.sum.Sum(nil)), c.err
}