package main

import (
	"context"
	"errors"
	"fmt"
	"os"
	"sync"

	"golang.org/x/sys/unix"
)

// openSyncer returns the syncer of the file system that holds dir. It syncs
// with syncfs, which writes back only that file system.
func openSyncer(dir string) (*Syncer, error) {
	file, err := os.Open(dir)
	if err != nil {
		return nil, fmt.Errorf("open chat directory: %w", err)
	}

	s := newSyncer(func() error {
		return unix.Syncfs(int(file.Fd()))
	})
	s.file = file

	return s, nil
}

// Syncer makes the writes of a chat durable by group commit. Each write goes to
// the operating system at once, so a crash of the process loses nothing. After
// each write, Written starts a background sync of the file system of the chat,
// unless one runs already. Then the next sync starts when the running one ends.
// One sync makes every write before its start durable, so a sync covers many
// writes, and there is never more than one sync at a time. A power loss loses
// at most the writes after the start of the last finished sync.
//
// A failed sync can mean that the kernel dropped written data, so the error
// stays, and each later Flush returns it.
type Syncer struct {
	mutex   sync.Mutex
	sync    func() error
	file    *os.File
	written uint64
	durable uint64
	running bool
	done    chan struct{}
	failure error
	workers sync.WaitGroup
}

// Written tells the syncer that a write went to the operating system, and
// starts a sync when none runs.
func (s *Syncer) Written() {
	s.mutex.Lock()
	defer s.mutex.Unlock()

	s.written++
	s.start()
}

// Flush returns when every write before the call is durable.
func (s *Syncer) Flush(ctx context.Context) error {
	s.mutex.Lock()
	target := s.written
	s.mutex.Unlock()

	for {
		s.mutex.Lock()
		reached := s.durable >= target
		failure := s.failure
		done := s.done
		s.mutex.Unlock()

		if reached {
			return nil
		}

		if failure != nil {
			return failure
		}

		select {
		case <-done:
		case <-ctx.Done():
			return ctx.Err()
		}
	}
}

// Close makes every write durable, waits for the syncs, and closes the
// directory.
func (s *Syncer) Close() error {
	err := s.Flush(context.Background())
	s.workers.Wait()
	if s.file != nil {
		err = errors.Join(err, s.file.Close())
		s.file = nil
	}

	return err
}

// start starts a sync of the writes so far when none runs and a write is not
// durable. The caller holds the mutex.
func (s *Syncer) start() {
	if s.running || s.durable == s.written || s.failure != nil {
		return
	}

	s.running = true
	s.workers.Add(1)
	go s.run(s.written)
}

// run syncs once, so the writes up to target become durable, and starts the
// next sync when writes came during this one.
func (s *Syncer) run(target uint64) {
	defer s.workers.Done()

	err := s.sync()

	s.mutex.Lock()
	defer s.mutex.Unlock()

	s.running = false
	if err != nil {
		s.failure = fmt.Errorf("sync chat files: %w", err)
	} else {
		s.durable = target
	}

	close(s.done)
	s.done = make(chan struct{})
	s.start()
}

// newSyncer returns a syncer that makes the writes durable with sync.
func newSyncer(sync func() error) *Syncer {
	return &Syncer{
		sync: sync,
		done: make(chan struct{}),
	}
}