dbbackup/internal/engine/binlog/streamer.go

// Package binlog provides MySQL binlog streaming capabilities for continuous backup.
// Uses native Go MySQL replication protocol for real-time binlog capture.
package binlog

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
	"sync"
	"sync/atomic"
	"time"
)

// Streamer handles continuous binlog streaming
type Streamer struct {
	config  *Config
	targets []Target
	state   *StreamerState
	log     Logger

	// Runtime state
	running   atomic.Bool
	stopCh    chan struct{}
	doneCh    chan struct{}
	mu        sync.RWMutex
	lastError error

	// Metrics
	eventsProcessed atomic.Uint64
	bytesProcessed  atomic.Uint64
	lastEventTime   atomic.Int64 // Unix timestamp
}

// Config contains binlog streamer configuration
type Config struct {
	// MySQL connection
	Host     string
	Port     int
	User     string
	Password string

	// Replication settings
	ServerID      uint32 // Must be unique in the replication topology
	Flavor        string // "mysql" or "mariadb"
	StartPosition *Position

	// Streaming mode
	Mode string // "continuous" or "oneshot"

	// Target configurations
	Targets []TargetConfig

	// Batching
	BatchMaxEvents int
	BatchMaxBytes  int
	BatchMaxWait   time.Duration

	// Checkpointing
	CheckpointEnabled  bool
	CheckpointFile     string
	CheckpointInterval time.Duration

	// Filtering
	Filter *Filter

	// GTID mode
	UseGTID bool
}

// TargetConfig contains target-specific configuration
type TargetConfig struct {
	Type string // "file", "s3", "kafka"

	// File target
	FilePath   string
	RotateSize int64

	// S3 target
	S3Bucket string
	S3Prefix string
	S3Region string

	// Kafka target
	KafkaBrokers []string
	KafkaTopic   string
}

// Position represents a binlog position
type Position struct {
	File     string `json:"file"`
	Position uint32 `json:"position"`
	GTID     string `json:"gtid,omitempty"`
}

// Filter defines what to include/exclude in streaming
type Filter struct {
	Databases        []string // Include only these databases (empty = all)
	Tables           []string // Include only these tables (empty = all)
	ExcludeDatabases []string // Exclude these databases
	ExcludeTables    []string // Exclude these tables
	Events           []string // Event types to include: "write", "update", "delete", "query"
	IncludeDDL       bool     // Include DDL statements
}

// StreamerState holds the current state of the streamer
type StreamerState struct {
	Position     Position       `json:"position"`
	EventCount   uint64         `json:"event_count"`
	ByteCount    uint64         `json:"byte_count"`
	LastUpdate   time.Time      `json:"last_update"`
	StartTime    time.Time      `json:"start_time"`
	TargetStatus []TargetStatus `json:"targets"`
}

// TargetStatus holds status for a single target
type TargetStatus struct {
	Name      string    `json:"name"`
	Type      string    `json:"type"`
	Healthy   bool      `json:"healthy"`
	LastWrite time.Time `json:"last_write"`
	Error     string    `json:"error,omitempty"`
}

// Event represents a parsed binlog event
type Event struct {
	Type      string           `json:"type"` // "write", "update", "delete", "query", "gtid", etc.
	Timestamp time.Time        `json:"timestamp"`
	Database  string           `json:"database,omitempty"`
	Table     string           `json:"table,omitempty"`
	Position  Position         `json:"position"`
	GTID      string           `json:"gtid,omitempty"`
	Query     string           `json:"query,omitempty"`    // For query events
	Rows      []map[string]any `json:"rows,omitempty"`     // For row events
	OldRows   []map[string]any `json:"old_rows,omitempty"` // For update events
	RawData   []byte           `json:"-"`                  // Raw binlog data for replay
	Extra     map[string]any   `json:"extra,omitempty"`
}

// Target interface for binlog output destinations
type Target interface {
	Name() string
	Type() string
	Write(ctx context.Context, events []*Event) error
	Flush(ctx context.Context) error
	Close() error
	Healthy() bool
}

// Logger interface for streamer logging
type Logger interface {
	Info(msg string, args ...any)
	Warn(msg string, args ...any)
	Error(msg string, args ...any)
	Debug(msg string, args ...any)
}

// NewStreamer creates a new binlog streamer
func NewStreamer(config *Config, log Logger) (*Streamer, error) {
	if config.ServerID == 0 {
		config.ServerID = 999 // Default server ID
	}
	if config.Flavor == "" {
		config.Flavor = "mysql"
	}
	if config.BatchMaxEvents == 0 {
		config.BatchMaxEvents = 1000
	}
	if config.BatchMaxBytes == 0 {
		config.BatchMaxBytes = 10 * 1024 * 1024 // 10MB
	}
	if config.BatchMaxWait == 0 {
		config.BatchMaxWait = 5 * time.Second
	}
	if config.CheckpointInterval == 0 {
		config.CheckpointInterval = 10 * time.Second
	}

	// Create targets
	targets := make([]Target, 0, len(config.Targets))
	for _, tc := range config.Targets {
		target, err := createTarget(tc)
		if err != nil {
			return nil, fmt.Errorf("failed to create target %s: %w", tc.Type, err)
		}
		targets = append(targets, target)
	}

	return &Streamer{
		config:  config,
		targets: targets,
		log:     log,
		state:   &StreamerState{StartTime: time.Now()},
		stopCh:  make(chan struct{}),
		doneCh:  make(chan struct{}),
	}, nil
}

// Start begins binlog streaming
func (s *Streamer) Start(ctx context.Context) error {
	if s.running.Swap(true) {
		return fmt.Errorf("streamer already running")
	}

	defer s.running.Store(false)
	defer close(s.doneCh)

	// Load checkpoint if exists
	if s.config.CheckpointEnabled {
		if err := s.loadCheckpoint(); err != nil {
			s.log.Warn("Could not load checkpoint, starting fresh", "error", err)
		}
	}

	s.log.Info("Starting binlog streamer",
		"host", s.config.Host,
		"port", s.config.Port,
		"server_id", s.config.ServerID,
		"mode", s.config.Mode,
		"targets", len(s.targets))

	// Use native Go implementation for binlog streaming
	return s.streamWithNative(ctx)
}

// streamWithNative uses pure Go MySQL protocol for streaming
func (s *Streamer) streamWithNative(ctx context.Context) error {
	// For production, we would use go-mysql-org/go-mysql library
	// This is a simplified implementation that polls SHOW BINARY LOGS
	// and reads binlog files incrementally

	// Start checkpoint goroutine
	if s.config.CheckpointEnabled {
		go s.checkpointLoop(ctx)
	}

	// Polling loop
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return s.shutdown()
		case <-s.stopCh:
			return s.shutdown()
		case <-ticker.C:
			if err := s.pollBinlogs(ctx); err != nil {
				s.log.Error("Error polling binlogs", "error", err)
				s.mu.Lock()
				s.lastError = err
				s.mu.Unlock()
			}
		}
	}
}

// pollBinlogs checks for new binlog data (simplified polling implementation)
func (s *Streamer) pollBinlogs(ctx context.Context) error {
	// In production, this would:
	// 1. Use MySQL replication protocol (COM_BINLOG_DUMP)
	// 2. Parse binlog events in real-time
	// 3. Call writeBatch() with parsed events

	// For now, this is a placeholder that simulates the polling
	// The actual implementation requires go-mysql-org/go-mysql

	return nil
}

// Stop stops the streamer gracefully
func (s *Streamer) Stop() error {
	if !s.running.Load() {
		return nil
	}

	close(s.stopCh)
	<-s.doneCh
	return nil
}

// shutdown performs cleanup
func (s *Streamer) shutdown() error {
	s.log.Info("Shutting down binlog streamer")

	// Flush all targets
	for _, target := range s.targets {
		if err := target.Flush(context.Background()); err != nil {
			s.log.Error("Error flushing target", "target", target.Name(), "error", err)
		}
		if err := target.Close(); err != nil {
			s.log.Error("Error closing target", "target", target.Name(), "error", err)
		}
	}

	// Save final checkpoint
	if s.config.CheckpointEnabled {
		s.saveCheckpoint()
	}

	return nil
}

// writeBatch writes a batch of events to all targets
func (s *Streamer) writeBatch(ctx context.Context, events []*Event) error {
	if len(events) == 0 {
		return nil
	}

	var lastErr error
	for _, target := range s.targets {
		if err := target.Write(ctx, events); err != nil {
			s.log.Error("Failed to write to target", "target", target.Name(), "error", err)
			lastErr = err
		}
	}

	// Update state
	last := events[len(events)-1]
	s.mu.Lock()
	s.state.Position = last.Position
	s.state.EventCount += uint64(len(events))
	s.state.LastUpdate = time.Now()
	s.mu.Unlock()

	s.eventsProcessed.Add(uint64(len(events)))
	s.lastEventTime.Store(last.Timestamp.Unix())

	return lastErr
}

// shouldProcess checks if an event should be processed based on filters
func (s *Streamer) shouldProcess(ev *Event) bool {
	if s.config.Filter == nil {
		return true
	}

	// Check database filter
	if len(s.config.Filter.Databases) > 0 {
		found := false
		for _, db := range s.config.Filter.Databases {
			if db == ev.Database {
				found = true
				break
			}
		}
		if !found {
			return false
		}
	}

	// Check exclude databases
	for _, db := range s.config.Filter.ExcludeDatabases {
		if db == ev.Database {
			return false
		}
	}

	// Check table filter
	if len(s.config.Filter.Tables) > 0 {
		found := false
		for _, t := range s.config.Filter.Tables {
			if t == ev.Table {
				found = true
				break
			}
		}
		if !found {
			return false
		}
	}

	// Check exclude tables
	for _, t := range s.config.Filter.ExcludeTables {
		if t == ev.Table {
			return false
		}
	}

	return true
}

// checkpointLoop periodically saves checkpoint
func (s *Streamer) checkpointLoop(ctx context.Context) {
	ticker := time.NewTicker(s.config.CheckpointInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-s.stopCh:
			return
		case <-ticker.C:
			s.saveCheckpoint()
		}
	}
}

// saveCheckpoint saves current position to file
func (s *Streamer) saveCheckpoint() error {
	if s.config.CheckpointFile == "" {
		return nil
	}

	s.mu.RLock()
	state := *s.state
	s.mu.RUnlock()

	data, err := json.MarshalIndent(state, "", "  ")
	if err != nil {
		return err
	}

	// Ensure directory exists
	if err := os.MkdirAll(filepath.Dir(s.config.CheckpointFile), 0755); err != nil {
		return err
	}

	// Write atomically
	tmpFile := s.config.CheckpointFile + ".tmp"
	if err := os.WriteFile(tmpFile, data, 0644); err != nil {
		return err
	}

	return os.Rename(tmpFile, s.config.CheckpointFile)
}

// loadCheckpoint loads position from checkpoint file
func (s *Streamer) loadCheckpoint() error {
	if s.config.CheckpointFile == "" {
		return nil
	}

	data, err := os.ReadFile(s.config.CheckpointFile)
	if err != nil {
		return err
	}

	var state StreamerState
	if err := json.Unmarshal(data, &state); err != nil {
		return err
	}

	s.mu.Lock()
	s.state = &state
	s.config.StartPosition = &state.Position
	s.mu.Unlock()

	s.log.Info("Loaded checkpoint",
		"file", state.Position.File,
		"position", state.Position.Position,
		"events", state.EventCount)

	return nil
}

// GetLag returns the replication lag
func (s *Streamer) GetLag() time.Duration {
	lastTime := s.lastEventTime.Load()
	if lastTime == 0 {
		return 0
	}
	return time.Since(time.Unix(lastTime, 0))
}

// Status returns current streamer status
func (s *Streamer) Status() *StreamerState {
	s.mu.RLock()
	defer s.mu.RUnlock()

	state := *s.state
	state.EventCount = s.eventsProcessed.Load()
	state.ByteCount = s.bytesProcessed.Load()

	// Update target status
	state.TargetStatus = make([]TargetStatus, 0, len(s.targets))
	for _, target := range s.targets {
		state.TargetStatus = append(state.TargetStatus, TargetStatus{
			Name:    target.Name(),
			Type:    target.Type(),
			Healthy: target.Healthy(),
		})
	}

	return &state
}

// Metrics returns streamer metrics
func (s *Streamer) Metrics() map[string]any {
	return map[string]any{
		"events_processed": s.eventsProcessed.Load(),
		"bytes_processed":  s.bytesProcessed.Load(),
		"lag_seconds":      s.GetLag().Seconds(),
		"running":          s.running.Load(),
	}
}

// createTarget creates a target based on configuration
func createTarget(tc TargetConfig) (Target, error) {
	switch tc.Type {
	case "file":
		return NewFileTarget(tc.FilePath, tc.RotateSize)
	case "s3":
		return NewS3Target(tc.S3Bucket, tc.S3Prefix, tc.S3Region)
	// case "kafka":
	// 	return NewKafkaTarget(tc.KafkaBrokers, tc.KafkaTopic)
	default:
		return nil, fmt.Errorf("unsupported target type: %s", tc.Type)
	}
}