core/cmd/twhelp/cmd_job.go

package main

import (
	"fmt"
	"log/slog"
	"slices"

	"gitea.dwysokinski.me/twhelp/core/internal/adapter"
	"gitea.dwysokinski.me/twhelp/core/internal/app"
	"github.com/urfave/cli/v2"
)

var (
	cmdJob = &cli.Command{
		Name:  "job",
		Usage: "Run various jobs (e.g. data sync)",
		Subcommands: []*cli.Command{
			{
				Name:  "sync",
				Usage: "Trigger various data syncs",
				Subcommands: []*cli.Command{
					{
						Name:  "data",
						Usage: "Trigger data sync (servers, players, tribes, villages)",
						Flags: slices.Concat(dbFlags, rmqFlags),
						Action: func(c *cli.Context) error {
							logger := loggerFromCtx(c.Context)
							watermillLogger := newWatermillLogger(logger)

							amqpConn, err := newAMQPConnectionFromFlags(c, watermillLogger)
							if err != nil {
								return err
							}
							defer func() {
								if closeErr := amqpConn.Close(); closeErr != nil {
									logger.Warn("couldn't close amqp connection", slog.Any("error", err))
								}
							}()

							publisher, err := newAMQPPublisher(amqpConn, watermillLogger)
							if err != nil {
								return err
							}
							defer func() {
								if closeErr := publisher.Close(); closeErr != nil {
									logger.Warn("couldn't close amqp publisher", slog.Any("error", err))
								}
							}()

							bunDB, err := newBunDBFromFlags(c)
							if err != nil {
								return err
							}
							defer closeBunDB(bunDB, logger)

							serverPublisher := adapter.NewServerWatermillPublisher(
								publisher,
								newWatermillMarshaler(),
								c.String(rmqFlagTopicSyncServersCmd.Name),
								c.String(rmqFlagTopicServerSyncedEvent.Name),
							)

							versionSvc := app.NewVersionService(adapter.NewVersionBunRepository(bunDB))
							dataSyncSvc := app.NewDataSyncService(versionSvc, nil, serverPublisher, nil)

							shutdownSignalCtx, stop := newShutdownSignalContext(c.Context)
							defer stop()

							if err = dataSyncSvc.Sync(shutdownSignalCtx); err != nil {
								return fmt.Errorf("couldn't trigger data sync: %w", err)
							}

							logger.Info("data sync triggered")

							return nil
						},
					},
					{
						Name:  "ennoblements",
						Usage: "Trigger ennoblement sync",
						Flags: slices.Concat(dbFlags, rmqFlags),
						Action: func(c *cli.Context) error {
							logger := loggerFromCtx(c.Context)
							watermillLogger := newWatermillLogger(logger)

							amqpConn, err := newAMQPConnectionFromFlags(c, watermillLogger)
							if err != nil {
								return err
							}
							defer func() {
								if closeErr := amqpConn.Close(); closeErr != nil {
									logger.Warn("couldn't close amqp connection", slog.Any("error", err))
								}
							}()

							publisher, err := newAMQPPublisher(amqpConn, watermillLogger)
							if err != nil {
								return err
							}
							defer func() {
								if closeErr := publisher.Close(); closeErr != nil {
									logger.Warn("couldn't close amqp publisher", slog.Any("error", err))
								}
							}()

							bunDB, err := newBunDBFromFlags(c)
							if err != nil {
								return err
							}
							defer closeBunDB(bunDB, logger)

							ennoblementPublisher := adapter.NewEnnoblementWatermillPublisher(
								publisher,
								newWatermillMarshaler(),
								c.String(rmqFlagTopicSyncEnnoblementsCmd.Name),
								c.String(rmqFlagTopicEnnoblementsSyncedEvent.Name),
							)

							versionSvc := app.NewVersionService(adapter.NewVersionBunRepository(bunDB))
							serverSvc := app.NewServerService(adapter.NewServerBunRepository(bunDB), nil, nil)
							dataSyncSvc := app.NewDataSyncService(versionSvc, serverSvc, nil, ennoblementPublisher)

							shutdownSignalCtx, stop := newShutdownSignalContext(c.Context)
							defer stop()

							if err = dataSyncSvc.SyncEnnoblements(shutdownSignalCtx); err != nil {
								return fmt.Errorf("couldn't trigger ennoblement sync: %w", err)
							}

							logger.Info("ennoblement sync triggered")

							return nil
						},
					},
				},
			},
			{
				Name: "create",
				Subcommands: []*cli.Command{
					{
						Name:  "snapshots",
						Usage: "Trigger snapshot creation (players/tribes)",
						Flags: slices.Concat(dbFlags, rmqFlags),
						Action: func(c *cli.Context) error {
							logger := loggerFromCtx(c.Context)
							watermillLogger := newWatermillLogger(logger)

							amqpConn, err := newAMQPConnectionFromFlags(c, watermillLogger)
							if err != nil {
								return err
							}
							defer func() {
								if closeErr := amqpConn.Close(); closeErr != nil {
									logger.Warn("couldn't close amqp connection", slog.Any("error", err))
								}
							}()

							publisher, err := newAMQPPublisher(amqpConn, watermillLogger)
							if err != nil {
								return err
							}
							defer func() {
								if closeErr := publisher.Close(); closeErr != nil {
									logger.Warn("couldn't close amqp publisher", slog.Any("error", err))
								}
							}()

							bunDB, err := newBunDBFromFlags(c)
							if err != nil {
								return err
							}
							defer closeBunDB(bunDB, logger)

							tribeSnapshotPublisher := adapter.NewSnapshotWatermillPublisher(
								publisher,
								newWatermillMarshaler(),
								c.String(rmqFlagTopicCreateTribeSnapshotsCmd.Name),
								c.String(rmqFlagTopicTribeSnapshotsCreatedEvent.Name),
							)

							playerSnapshotPublisher := adapter.NewSnapshotWatermillPublisher(
								publisher,
								newWatermillMarshaler(),
								c.String(rmqFlagTopicCreatePlayerSnapshotsCmd.Name),
								c.String(rmqFlagTopicPlayerSnapshotsCreatedEvent.Name),
							)

							versionSvc := app.NewVersionService(adapter.NewVersionBunRepository(bunDB))
							serverSvc := app.NewServerService(adapter.NewServerBunRepository(bunDB), nil, nil)
							snapshotSvc := app.NewSnapshotService(versionSvc, serverSvc, tribeSnapshotPublisher, playerSnapshotPublisher)

							shutdownSignalCtx, stop := newShutdownSignalContext(c.Context)
							defer stop()

							if err = snapshotSvc.Create(shutdownSignalCtx); err != nil {
								return fmt.Errorf("couldn't trigger snapshot creation: %w", err)
							}

							logger.Info("snapshot creation triggered")

							return nil
						},
					},
				},
			},
			{
				Name:        "cleanup",
				Description: "Trigger data cleanup",
				Flags:       slices.Concat(dbFlags, rmqFlags),
				Action: func(c *cli.Context) error {
					logger := loggerFromCtx(c.Context)
					watermillLogger := newWatermillLogger(logger)

					amqpConn, err := newAMQPConnectionFromFlags(c, watermillLogger)
					if err != nil {
						return err
					}
					defer func() {
						if closeErr := amqpConn.Close(); closeErr != nil {
							logger.Warn("couldn't close amqp connection", slog.Any("error", err))
						}
					}()

					publisher, err := newAMQPPublisher(amqpConn, watermillLogger)
					if err != nil {
						return err
					}
					defer func() {
						if closeErr := publisher.Close(); closeErr != nil {
							logger.Warn("couldn't close amqp publisher", slog.Any("error", err))
						}
					}()

					bunDB, err := newBunDBFromFlags(c)
					if err != nil {
						return err
					}
					defer closeBunDB(bunDB, logger)

					dataCleanupPublisher := adapter.NewDataCleanupPublisher(
						publisher,
						newWatermillMarshaler(),
						c.String(rmqFlagTopicCleanUpDataCmd.Name),
					)

					serverSvc := app.NewServerService(adapter.NewServerBunRepository(bunDB), nil, nil)
					dataCleanupSvc := app.NewDataCleanupService(serverSvc, dataCleanupPublisher)

					shutdownSignalCtx, stop := newShutdownSignalContext(c.Context)
					defer stop()

					if err = dataCleanupSvc.CleanUp(shutdownSignalCtx); err != nil {
						return fmt.Errorf("couldn't trigger data cleanup: %w", err)
					}

					logger.Info("data cleanup triggered")

					return nil
				},
			},
		},
	}
)