100 lines
2.9 KiB
Go
100 lines
2.9 KiB
Go
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
|
|
}
|