glpi-neural-brain/internal/persist/coordinator.go

package persist

import (
	"context"
	"fmt"
	"os"
	"path/filepath"
	"sort"
	"sync"
	"time"

	"github.com/local/glpi-neural-brain/internal/activity"
	"github.com/local/glpi-neural-brain/internal/graph"
	"github.com/local/glpi-neural-brain/internal/model"
)

type pendingFile struct {
	path string
	data []byte
	perm os.FileMode
}

type Status struct {
	Interval             string    `json:"interval"`
	PendingFiles         int       `json:"pending_files"`
	GraphDirty           bool      `json:"graph_dirty"`
	LastFlush            time.Time `json:"last_flush,omitempty"`
	LastError            string    `json:"last_error,omitempty"`
	LastPersistedVersion uint64    `json:"last_persisted_version"`
	CurrentGraphVersion  uint64    `json:"current_graph_version"`
	SuccessfulFlushes    uint64    `json:"successful_flushes"`
	FailedFlushes        uint64    `json:"failed_flushes"`
}

type Coordinator struct {
	graph    *graph.Store
	broker   *activity.Broker
	interval time.Duration

	mu                   sync.RWMutex
	files                map[string]pendingFile
	lastFlush            time.Time
	lastError            string
	lastPersistedVersion uint64
	successfulFlushes    uint64
	failedFlushes        uint64
	flushMu              sync.Mutex
}

func New(g *graph.Store, b *activity.Broker, interval time.Duration) *Coordinator {
	if interval <= 0 {
		interval = 5 * time.Minute
	}
	version := uint64(0)
	if g != nil {
		version = g.Version()
	}
	return &Coordinator{graph: g, broker: b, interval: interval, files: make(map[string]pendingFile), lastPersistedVersion: version}
}

func (c *Coordinator) Start(ctx context.Context) {
	go func() {
		ticker := time.NewTicker(c.interval)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				if err := c.Flush(ctx, "interval"); err != nil && c.broker != nil {
					c.broker.Publish(model.Activity{Type: "persistence.failed", Source: "brain", Phase: "storage", Message: "Gebündeltes Schreiben ist fehlgeschlagen", Strength: .35, Metadata: map[string]any{"error": err.Error()}})
				}
			}
		}
	}()
}

func (c *Coordinator) QueueFile(path string, data []byte, perm os.FileMode) (string, error) {
	if path == "" {
		return "", fmt.Errorf("persistence path is empty")
	}
	if perm == 0 {
		perm = 0o640
	}
	abs, err := filepath.Abs(path)
	if err != nil {
		return "", err
	}
	c.mu.Lock()
	c.files[abs] = pendingFile{path: abs, data: append([]byte(nil), data...), perm: perm}
	c.mu.Unlock()
	return abs, nil
}

func (c *Coordinator) Pending(path string) bool {
	abs, err := filepath.Abs(path)
	if err != nil {
		return false
	}
	c.mu.RLock()
	_, ok := c.files[abs]
	c.mu.RUnlock()
	return ok
}

func (c *Coordinator) Flush(ctx context.Context, trigger string) error {
	c.flushMu.Lock()
	defer c.flushMu.Unlock()

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

	c.mu.RLock()
	files := make([]pendingFile, 0, len(c.files))
	for _, f := range c.files {
		files = append(files, pendingFile{path: f.path, data: append([]byte(nil), f.data...), perm: f.perm})
	}
	lastVersion := c.lastPersistedVersion
	c.mu.RUnlock()
	sort.Slice(files, func(i, j int) bool { return files[i].path < files[j].path })

	currentVersion := lastVersion
	if c.graph != nil {
		currentVersion = c.graph.Version()
	}
	graphDirty := currentVersion != lastVersion
	if c.graph != nil {
		graphDirty = c.graph.Dirty()
	}
	if len(files) == 0 && !graphDirty {
		return nil
	}

	// Knowledge/staging/cache files are committed first. The incremental SQLite
	// transaction follows so a persisted graph never points at a draft that has
	// not yet reached disk.
	for _, f := range files {
		if err := writeAtomic(f.path, f.data, f.perm); err != nil {
			c.recordFailure(err)
			return err
		}
		c.mu.Lock()
		if pending, ok := c.files[f.path]; ok && string(pending.data) == string(f.data) {
			delete(c.files, f.path)
		}
		c.mu.Unlock()
	}

	if graphDirty && c.graph != nil {
		persistedVersion, err := c.graph.PersistVersionContext(ctx)
		if err != nil {
			c.recordFailure(err)
			return err
		}
		currentVersion = persistedVersion
	}

	now := time.Now().UTC()
	c.mu.Lock()
	c.lastFlush = now
	c.lastError = ""
	c.lastPersistedVersion = currentVersion
	c.successfulFlushes++
	remaining := len(c.files)
	c.mu.Unlock()
	if c.broker != nil {
		c.broker.Publish(model.Activity{Type: "persistence.flushed", Source: "brain", Phase: "storage", Message: "SQLite-Graph und Wissensdateien wurden gebündelt geschrieben", Strength: .32, Metadata: map[string]any{"trigger": trigger, "files": len(files), "graph_version": currentVersion, "remaining_files": remaining}})
	}
	return nil
}

func (c *Coordinator) recordFailure(err error) {
	c.mu.Lock()
	c.lastError = err.Error()
	c.failedFlushes++
	c.mu.Unlock()
}

func (c *Coordinator) Status() Status {
	current := uint64(0)
	if c.graph != nil {
		current = c.graph.Version()
	}
	c.mu.RLock()
	defer c.mu.RUnlock()
	return Status{
		Interval:     c.interval.String(),
		PendingFiles: len(c.files),
		GraphDirty: func() bool {
			if c.graph != nil {
				return c.graph.Dirty()
			}
			return current != c.lastPersistedVersion
		}(),
		LastFlush:            c.lastFlush,
		LastError:            c.lastError,
		LastPersistedVersion: c.lastPersistedVersion,
		CurrentGraphVersion:  current,
		SuccessfulFlushes:    c.successfulFlushes,
		FailedFlushes:        c.failedFlushes,
	}
}

func writeAtomic(path string, data []byte, perm os.FileMode) error {
	if err := os.MkdirAll(filepath.Dir(path), 0o750); err != nil {
		return err
	}
	tmp, err := os.CreateTemp(filepath.Dir(path), ".brain-write-*")
	if err != nil {
		return err
	}
	tmpName := tmp.Name()
	defer os.Remove(tmpName)
	if err := tmp.Chmod(perm); err != nil {
		tmp.Close()
		return err
	}
	if _, err := tmp.Write(data); err != nil {
		tmp.Close()
		return err
	}
	if err := tmp.Sync(); err != nil {
		tmp.Close()
		return err
	}
	if err := tmp.Close(); err != nil {
		return err
	}
	return os.Rename(tmpName, path)
}