package actor

import (
	"context"
	"errors"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"google.golang.org/protobuf/proto"
)

const defaultWriterIdleTTL = 5 * time.Minute

type QueueEvent struct {
	TimeStamp time.Time
	Message   proto.Message
}

// diskLogWriter holds one append-only log file kept open between writes.
type diskLogWriter struct {
	mu       sync.Mutex
	file     *os.File
	lastUsed time.Time
	purged   bool // set by purgeIdleWriters before closing; getWriter re-checks after acquiring mu
}

type DiskStorage[V any] struct {
	*StateStorage
	path          string
	done          chan struct{}
	queue         *sync.Map // map[uint64][]QueueEvent
	dirOnce       sync.Once
	dirErr        error
	writersMu     sync.Mutex
	writers       map[uint64]*diskLogWriter
	writerIdleTTL time.Duration
}

type LogStorage[V any] interface {
	LoadEvents(ctx context.Context, id uint64, grain Grain[V]) error
	LoadEventsFunc(ctx context.Context, id uint64, grain Grain[V], condition func(msg proto.Message, index int, timeStamp time.Time) bool) error
	AppendMutations(id uint64, msg ...proto.Message) error
}

func NewDiskStorage[V any](path string, registry MutationRegistry) *DiskStorage[V] {
	s := &DiskStorage[V]{
		StateStorage:  NewState(registry),
		path:          path,
		done:          make(chan struct{}),
		writers:       make(map[uint64]*diskLogWriter),
		writerIdleTTL: defaultWriterIdleTTL,
	}
	go s.purgeLoop()
	go s.filesExistingLoop()
	return s
}

// filesExistingLoop refreshes the EventLogFilesExisting gauge every
// 30 seconds. The gauge represents the count of .events.log files on
// disk, which diverges from the open-writer count whenever the idle
// purge closes a writer but the file is still kept (the .log file is
// not deleted by the idle purge). Periodic re-listing is cheap (one
// os.ReadDir per tick) and is the only way the gauge can pick up files
// created or removed out of band (e.g. GDPR purge, or the init-time
// rebuild of a grain that already has a log).
func (s *DiskStorage[V]) filesExistingLoop() {
	ticker := time.NewTicker(30 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-s.done:
			return
		case <-ticker.C:
			s.refreshFilesExisting()
		}
	}
}

// refreshFilesExisting counts .events.log files in the storage dir
// and writes the value to the EventLogFilesExisting gauge. Safe to
// call concurrently — os.ReadDir is atomic for our purposes (the dir
// is owned by this pod, mutations go through the writersMu lock).
// No-op when the dir does not exist yet (cold start) or when metrics
// are disabled (s.StateStorage.metrics == nil).
func (s *DiskStorage[V]) refreshFilesExisting() {
	if s.StateStorage.metrics == nil {
		return
	}
	entries, err := os.ReadDir(s.path)
	if err != nil {
		// Dir may not exist yet on a cold start, or briefly during a
		// retention-driven directory recreation. Leave the gauge
		// untouched in that case — the next tick will catch up.
		return
	}
	count := 0
	for _, e := range entries {
		if !e.IsDir() && strings.HasSuffix(e.Name(), ".events.log") {
			count++
		}
	}
	s.StateStorage.metrics.EventLogFilesExisting.Set(float64(count))
}

func (s *DiskStorage[V]) ensureDir() error {
	s.dirOnce.Do(func() {
		s.dirErr = os.MkdirAll(s.path, 0o755)
	})
	return s.dirErr
}

// getWriter returns an open append handle for id. The caller must unlock w.mu
// when finished writing.


// SetMetrics attaches a Metrics instance to the embedded StateStorage
// so every append / replay / unknown-type path has access to the
// same event-log counters. The metrics struct is shared by reference;
// do not mutate its fields from outside. Pass nil to disable
// instrumentation (used by tests that do not care about Prometheus).
func (s *DiskStorage[V]) SetMetrics(m *Metrics) {
	s.StateStorage.metrics = m
}

func (s *DiskStorage[V]) recordAppend(n int) {
	m := s.StateStorage.metrics
	if m == nil {
		return
	}
	m.EventLogAppends.Inc()
	if n > 0 {
		m.EventLogBytesWritten.Add(float64(n))
	}
	m.EventLogLastAppendUnix.Set(float64(time.Now().Unix()))
}

func (s *DiskStorage[V]) writeEvents(id uint64, events []QueueEvent) error {
	if len(events) == 0 {
		return nil
	}
	w, err := s.getWriter(id)
	if err != nil {
		return err
	}
	defer w.mu.Unlock()

	for _, evt := range events {
		n, err := s.Append(w.file, evt.Message, evt.TimeStamp)
		if err != nil {
			// A partial write still counts the bytes that DID land on
			// disk, so the throughput gauge is honest about a mid-write
			// failure.
			s.recordAppend(n)
			return err
		}
		s.recordAppend(n)
	}
	w.lastUsed = time.Now()
	return nil
}

func (s *DiskStorage[V]) writeMutations(id uint64, msgs ...proto.Message) error {
	if len(msgs) == 0 {
		return nil
	}
	w, err := s.getWriter(id)
	if err != nil {
		return err
	}
	defer w.mu.Unlock()

	now := time.Now()
	for _, m := range msgs {
		n, err := s.Append(w.file, m, now)
		if err != nil {
			s.recordAppend(n)
			return err
		}
		s.recordAppend(n)
	}
	w.lastUsed = time.Now()
	return nil
}

func (s *DiskStorage[V]) purgeLoop() {
	ticker := time.NewTicker(time.Minute)
	defer ticker.Stop()
	for {
		select {
		case <-s.done:
			return
		case <-ticker.C:
			s.purgeIdleWriters()
		}
	}
}

func (s *DiskStorage[V]) purgeIdleWriters() {
	cutoff := time.Now().Add(-s.writerIdleTTL)

	s.writersMu.Lock()
	defer s.writersMu.Unlock()

	for id, w := range s.writers {
		w.mu.Lock()
		if w.lastUsed.Before(cutoff) {
			w.purged = true
			delete(s.writers, id)
			if err := w.file.Close(); err != nil {
				log.Printf("failed to close idle event log %d: %v", id, err)
			}
		}
		w.mu.Unlock()
	}
}

func (s *DiskStorage[V]) getWriter(id uint64) (*diskLogWriter, error) {
	if err := s.ensureDir(); err != nil {
		return nil, err
	}

	for {
		s.writersMu.Lock()
		w, ok := s.writers[id]
		if !ok {
			fh, err := os.OpenFile(s.logPath(id), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
			if err != nil {
				s.writersMu.Unlock()
				return nil, err
			}
			w = &diskLogWriter{file: fh, lastUsed: time.Now()}
			s.writers[id] = w
		}
		s.writersMu.Unlock()

		w.mu.Lock()
		if !w.purged {
			w.lastUsed = time.Now()
			return w, nil
		}
		// Writer was purged by purgeIdleWriters between our map lookup and acquiring w.mu.
		// Release it and retry — the map no longer contains this entry, so the next
		// iteration will create a fresh writer.
		w.mu.Unlock()
	}
}

func (s *DiskStorage[V]) closeWriters() {
	s.writersMu.Lock()
	defer s.writersMu.Unlock()
	for id, w := range s.writers {
		w.mu.Lock()
		w.purged = true
		if err := w.file.Close(); err != nil {
			log.Printf("failed to close event log %d: %v", id, err)
		}
		delete(s.writers, id)
		w.mu.Unlock()
	}
}

func (s *DiskStorage[V]) SaveLoop(duration time.Duration) {
	s.queue = &sync.Map{}
	ticker := time.NewTicker(duration)
	defer ticker.Stop()
	for {
		select {
		case <-s.done:
			s.save()
			return
		case <-ticker.C:
			s.save()
		}
	}
}

func (s *DiskStorage[V]) save() {
	if s.queue == nil {
		return
	}
	carts := 0
	lines := 0
	s.queue.Range(func(key, value any) bool {
		id := key.(uint64)
		qe, ok := value.([]QueueEvent)
		if !ok {
			s.queue.Delete(id)
			return true
		}
		if err := s.writeEvents(id, qe); err != nil {
			log.Printf("failed to append events for grain %d: %v", id, err)
			return true
		}
		lines += len(qe)
		carts++
		s.queue.Delete(id)
		return true
	})
	if lines > 0 {
		log.Printf("Appended %d carts and %d lines to disk", carts, lines)
	}
}

func (s *DiskStorage[V]) logPath(id uint64) string {
	return filepath.Join(s.path, fmt.Sprintf("%d.events.log", id))
}

// loadReplayMetrics times the replay, records duration, and tracks
// handler errors. It is shared by LoadEvents and LoadEventsFunc. The
// returned `onMessage` closure bumps EventLogMutationErrors on any
// non-nil error from registry.Apply, and the deferred function bumps
// EventLogReplayTotal / Failures / Duration on the overall return.
func (s *DiskStorage[V]) loadReplayMetrics() (onMessage func(err error), done func(err error)) {
	m := s.StateStorage.metrics
	if m == nil {
		noop := func(error) {}
		return noop, noop
	}
	start := time.Now()
	onMessage = func(err error) {
		if err != nil {
			m.EventLogMutationErrors.Inc()
		}
	}
	done = func(err error) {
		if err != nil {
			m.EventLogReplayFailures.Inc()
			return
		}
		m.EventLogReplayTotal.Inc()
		m.EventLogReplayDuration.Observe(time.Since(start).Seconds())
	}
	return onMessage, done
}

// recordReplayFailure bumps EventLogReplayFailures when the open
// step fails before the Load loop runs (and therefore before the
// loadReplayMetrics closure is in scope). Nil-safe.
func (s *DiskStorage[V]) recordReplayFailure() {
	if m := s.StateStorage.metrics; m != nil {
		m.EventLogReplayFailures.Inc()
	}
}

func (s *DiskStorage[V]) LoadEventsFunc(ctx context.Context, id uint64, grain Grain[V], condition func(msg proto.Message, index int, timeStamp time.Time) bool) error {
	path := s.logPath(id)
	if _, err := os.Stat(path); errors.Is(err, os.ErrNotExist) {
		// No log -> nothing to replay
		return nil
	}

	fh, err := os.Open(path)
	if err != nil {
		s.recordReplayFailure()
		return fmt.Errorf("open replay file: %w", err)
	}
	defer fh.Close()
	trackApply, done := s.loadReplayMetrics()
	index := 0
	loadErr := s.Load(fh, func(msg proto.Message, when time.Time) {
		if condition(msg, index, when) {
			results, _ := s.registry.Apply(ctx, grain, msg)
			for _, r := range results {
				if r.Error != nil {
					trackApply(r.Error)
				}
			}
		}
		index++
	})
	done(loadErr)
	return loadErr
}

func (s *DiskStorage[V]) LoadEvents(ctx context.Context, id uint64, grain Grain[V]) error {
	path := s.logPath(id)
	if _, err := os.Stat(path); errors.Is(err, os.ErrNotExist) {
		// No log -> nothing to replay
		return nil
	}

	fh, err := os.Open(path)
	if err != nil {
		s.recordReplayFailure()
		return fmt.Errorf("open replay file: %w", err)
	}
	defer fh.Close()
	trackApply, done := s.loadReplayMetrics()
	loadErr := s.Load(fh, func(msg proto.Message, _ time.Time) {
		results, _ := s.registry.Apply(ctx, grain, msg)
		for _, r := range results {
			if r.Error != nil {
				trackApply(r.Error)
			}
		}
	})
	done(loadErr)
	return loadErr
}

func (s *DiskStorage[V]) Close() {
	if s.queue != nil {
		s.save()
	}
	s.closeWriters()
	close(s.done)
}

func (s *DiskStorage[V]) AppendMutations(id uint64, msg ...proto.Message) error {
	if s.queue != nil {
		queue := make([]QueueEvent, 0)
		data, found := s.queue.Load(id)
		if found {
			queue = data.([]QueueEvent)
		}
		for _, m := range msg {
			queue = append(queue, QueueEvent{Message: m, TimeStamp: time.Now()})
		}
		s.queue.Store(id, queue)
		return nil
	}
	return s.writeMutations(id, msg...)
}

func (s *DiskStorage[V]) openWriterCount() int {
	s.writersMu.Lock()
	defer s.writersMu.Unlock()
	return len(s.writers)
}

func (s *DiskStorage[V]) PurgeOlderThan(ctx context.Context, retention time.Duration, isGrainActive func(id uint64) bool) (int, error) {
	if err := s.ensureDir(); err != nil {
		return 0, err
	}

	files, err := os.ReadDir(s.path)
	if err != nil {
		return 0, fmt.Errorf("read storage dir: %w", err)
	}

	cutoff := time.Now().Add(-retention)
	purgedCount := 0

	for _, entry := range files {
		if entry.IsDir() {
			continue
		}
		name := entry.Name()
		if !strings.HasSuffix(name, ".events.log") {
			continue
		}

		var id uint64
		_, err := fmt.Sscanf(name, "%d.events.log", &id)
		if err != nil {
			continue
		}

		info, err := entry.Info()
		if err != nil {
			log.Printf("failed to get file info for %s: %v", name, err)
			continue
		}

		if info.ModTime().Before(cutoff) {
			// Check if grain is active in memory
			if isGrainActive != nil && isGrainActive(id) {
				continue
			}

			// Check if we have an active writer open
			s.writersMu.Lock()
			_, hasWriter := s.writers[id]
			s.writersMu.Unlock()
			if hasWriter {
				continue
			}

			// Delete the file
			fullPath := filepath.Join(s.path, name)
			if err := os.Remove(fullPath); err != nil {
				log.Printf("failed to delete expired cart log %s: %v", fullPath, err)
			} else {
				purgedCount++
			}
		}
	}

	return purgedCount, nil
}