bybit_btcusdt_ingest/main.go

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"
)

func main() {
	log.SetFlags(log.LstdFlags | log.Lmicroseconds)

	// Load configuration
	cfg, err := LoadConfig("config.json")
	if err != nil {
		log.Fatalf("Config error: %v", err)
	}

	if len(os.Args) < 2 {
		printUsage()
		return
	}

	command := os.Args[1]

	switch command {
	case "run":
		runEngine(cfg)

	case "recover":
		log.Println("=== Starting Recovery Mode ===")
		sm, err := NewStorageManager(cfg)
		if err != nil {
			log.Fatal(err)
		}
		if err := sm.StartupRecovery(); err != nil {
			log.Fatal(err)
		}
		log.Println("Recovery completed.")

	case "maintain":
		log.Println("=== Starting Manual Maintenance ===")
		sm, err := NewStorageManager(cfg)
		if err != nil {
			log.Fatal(err)
		}
		sm.RunHourlyMaintenance()
		log.Println("Maintenance completed.")

	case "help":
		printUsage()

	default:
		log.Printf("Unknown command: %s", command)
		printUsage()
	}
}

func printUsage() {
	fmt.Println("Usage: engine [command]")
	fmt.Println("\nCommands:")
	fmt.Println("  run       Start the WebSocket ingestor and processing engine")
	fmt.Println("  recover   Run startup recovery to migrate stale ticks")
	fmt.Println("  maintain  Run a single maintenance cycle (cleanup/retention)")
	fmt.Println("  help      Show this help message")
}

func runEngine(cfg Config) {
	log.Println("=== Bybit BTC/USDT Tick Ingest Engine ===")
	log.Printf("Config: symbol=%s, hot_retention=%dh, feature_retention=%dd",
		cfg.Symbol, cfg.HotRetentionHours, cfg.FeatureRetentionDays)

	// Initialize storage (creates dirs, databases, tables)
	sm, err := NewStorageManager(cfg)
	if err != nil {
		log.Fatalf("Storage init error: %v", err)
	}

	// Startup recovery: migrate any stale ticks from previous runs
	if err := sm.StartupRecovery(); err != nil {
		log.Fatalf("Startup recovery error: %v", err)
	}

	// Create the tick channel (buffered to absorb bursts)
	tickCh := make(chan Tick, cfg.TickChannelBuffer)

	// Initialize aggregator (5-second feature bucketing)
	agg, err := NewAggregator(sm)
	if err != nil {
		log.Fatalf("Aggregator init error: %v", err)
	}

	// Initialize batch writer
	writer, err := NewWriter(sm, tickCh, cfg)
	if err != nil {
		log.Fatalf("Writer init error: %v", err)
	}

	// Create ingestor
	ingestor := NewIngestor(cfg, tickCh, agg)

	// Context for graceful shutdown
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Catch OS signals
	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)

	var wg sync.WaitGroup

	// Goroutine 1: Batch Writer
	wg.Add(1)
	go func() {
		defer wg.Done()
		writer.Run(ctx)
	}()

	// Goroutine 2: WebSocket Ingestor
	wg.Add(1)
	go func() {
		defer wg.Done()
		ingestor.Run(ctx)
	}()

	// Goroutine 3: Hourly Maintenance
	wg.Add(1)
	go func() {
		defer wg.Done()
		maintInterval := time.Duration(cfg.MaintenanceIntervalMinutes) * time.Minute
		ticker := time.NewTicker(maintInterval)
		defer ticker.Stop()

		log.Printf("[maintenance] Scheduled every %d minutes.", cfg.MaintenanceIntervalMinutes)
		for {
			select {
			case <-ctx.Done():
				log.Println("[maintenance] Maintenance goroutine stopped.")
				return
			case <-ticker.C:
				sm.RunHourlyMaintenance()
			}
		}
	}()

	// Wait for shutdown signal
	sig := <-sigCh
	log.Printf("Received signal %v, initiating graceful shutdown...", sig)
	cancel()

	// Close the tick channel so the writer drains remaining ticks
	close(tickCh)

	// Flush any remaining aggregator bucket
	agg.Close()

	// Wait for all goroutines to finish
	wg.Wait()
	log.Println("=== Engine shut down cleanly. ===")
}