package main

import (
	"context"
	"errors"
	"fmt"
	"log"
	"os"
	"os/signal"
	"strconv"
	"strings"
	"syscall"
	"time"

	"gitea.kazu634.com/kazu634/tver/internal/store"
	"gitea.kazu634.com/kazu634/tver/internal/tver"
	workerpkg "gitea.kazu634.com/kazu634/tver/internal/worker"
	"github.com/jackc/pgx/v5/pgxpool"
)

const staleAfter = 30 * time.Second

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()
	config, err := loadConfig(os.Getenv)
	if err != nil {
		log.Fatal(err)
	}
	pool, err := initializeDatabase(ctx, os.Getenv("DATABASE_URL"), time.Now(), pgxpool.New,
		store.Migrate, func(ctx context.Context, pool *pgxpool.Pool, cutoff time.Time) (int64, error) {
			return store.New(pool).RequeueStale(ctx, cutoff)
		})
	if err != nil {
		log.Fatal(err)
	}
	defer pool.Close()
	jobStore := store.New(pool)
	runner, err := workerpkg.New(jobStore, config, log.Default())
	if err != nil {
		log.Fatal(err)
	}
	tverClient, err := tver.NewClient("", nil)
	if err != nil {
		log.Fatal(err)
	}
	scheduler, err := workerpkg.NewSubscriptionScheduler(jobStore, tverClient,
		workerpkg.SubscriptionSchedulerConfig{PreviewMaxSeconds: tver.PreviewMaxSeconds}, log.Default())
	if err != nil {
		log.Fatal(err)
	}
	schedulerDone := make(chan struct{})
	go func() {
		defer close(schedulerDone)
		scheduler.Run(ctx)
	}()
	log.Printf("worker started (concurrency=%d)", config.Concurrency)
	if err := runner.Run(ctx); err != nil {
		log.Fatal(err)
	}
	<-schedulerDone
	log.Print("worker stopped")
}

func loadConfig(getenv func(string) string) (workerpkg.Config, error) {
	concurrency := 1
	if value := strings.TrimSpace(getenv("WORKER_CONCURRENCY")); value != "" {
		parsed, err := strconv.Atoi(value)
		if err != nil || parsed < 1 || parsed > 2 {
			return workerpkg.Config{}, errors.New("WORKER_CONCURRENCY は 1 または 2 を指定してください")
		}
		concurrency = parsed
	}
	return workerpkg.Config{VideoDir: getenv("VIDEO_DIR"), YTDLPPath: getenv("YT_DLP_PATH"),
		FFprobePath: getenv("FFPROBE_PATH"), Concurrency: concurrency}, nil
}

func initializeDatabase(ctx context.Context, databaseURL string, now time.Time,
	connect func(context.Context, string) (*pgxpool.Pool, error),
	migrate func(context.Context, *pgxpool.Pool) error,
	requeue func(context.Context, *pgxpool.Pool, time.Time) (int64, error),
) (*pgxpool.Pool, error) {
	if strings.TrimSpace(databaseURL) == "" {
		return nil, errors.New("DATABASE_URL が設定されていません")
	}
	pool, err := connect(ctx, databaseURL)
	if err != nil {
		return nil, errors.New("DATABASE_URL から接続プールを作成できません")
	}
	if err := migrate(ctx, pool); err != nil {
		pool.Close()
		return nil, fmt.Errorf("DB マイグレーションに失敗: %w", err)
	}
	if _, err := requeue(ctx, pool, now.Add(-staleAfter)); err != nil {
		pool.Close()
		return nil, fmt.Errorf("停止中ジョブの復帰に失敗: %w", err)
	}
	return pool, nil
}