job-search/internal/cli/serve/command.go

package serve

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

	"github.com/spf13/cobra"

	"git.avsdev.uk/Go/job-search/internal/cli"
	"git.avsdev.uk/Go/job-search/internal/config"
	"git.avsdev.uk/Go/job-search/internal/events"
	"git.avsdev.uk/Go/job-search/internal/fetcher"
	"git.avsdev.uk/Go/job-search/internal/frontend"
	"git.avsdev.uk/Go/job-search/internal/logging"
	"git.avsdev.uk/Go/job-search/internal/server"
)

// NewCommand returns the serve command, which serves the REST API, the
// WebSocket endpoint and the frontend, and runs a worker for every enabled
// source.
func NewCommand(opts *cli.Options) *cobra.Command {
	cmd := &cobra.Command{
		Use:   "serve",
		Short: "Serve the REST API, WebSocket and frontend, and fetch from sources",
		Long: `Serve the REST API under /api, the WebSocket endpoint under /ws, and the
frontend's files from server.frontend.dir under / if it is set.

It also runs a worker for every enabled source, as fetcher service does,
saving what they fetch to database.file if it is set, and pushing an
OpportunitiesUpdated event to every WebSocket client when a save stores new
opportunities or new releases. A source that stops with an error stops the
server too, and the server stopping stops the sources. With database.file
set, the API serves the stored opportunities under /api/opportunities.`,
		PersistentPreRun: func(cmd *cobra.Command, args []string) {
			slog.Debug("command persistent pre-run", "command", "serve")
		},
		PersistentPostRun: func(cmd *cobra.Command, args []string) {
			slog.Debug("command persistent post-run", "command", "serve")
		},
		Args: cobra.NoArgs,
		RunE: func(cmd *cobra.Command, _ []string) (err error) {
			slog.Debug("command run", "command", "serve")

			addrOverride, _ := cmd.Flags().GetString("listen")
			hostOverride, _ := cmd.Flags().GetString("host")
			portOverride, _ := cmd.Flags().GetInt("port")
			foreground, _ := cmd.Flags().GetBool("foreground")
			accessLogOverride, _ := cmd.Flags().GetString("access-log")
			since, _ := cmd.Flags().GetTime("since")

			if addrOverride != "" {
				listenParts := strings.Split(addrOverride, ":")
				if len(listenParts) != 2 {
					return errors.New(
						"--listen should be in the format of 127.0.0.1:9393",
					)
				}
				portOverride, err = strconv.Atoi(listenParts[1])
				if err != nil {
					return fmt.Errorf("failed to parse --listen port: %w", err)
				}
				hostOverride = listenParts[0]
			}

			conf, err := opts.RunSetup(cmd, true, foreground)
			if err != nil {
				return fmt.Errorf("load configuration: %w", err)
			}

			if portOverride != 0 {
				conf.Server.Port = portOverride
			}
			if hostOverride != "" {
				conf.Server.Host = hostOverride
			}
			if cmd.Flags().Changed("access-log") {
				conf.Server.AccessLog = accessLogOverride
			}

			options, closeAccessLog, err := serverOptions(&conf.Server)
			if err != nil {
				return err
			}
			defer func() {
				err = errors.Join(err, closeAccessLog())
			}()

			// ctx is cancelled on the first SIGINT (Ctrl+C) or SIGTERM.
			ctx, stop := signal.NotifyContext(
				context.Background(), os.Interrupt, syscall.SIGTERM,
			)
			defer stop()

			// After the first signal, restore default signal handling so a second
			// Ctrl+C force-kills the process if graceful shutdown hangs.
			go func() {
				<-ctx.Done()
				stop()
			}()

			// The store is shared by the sources, which save to it, and the API,
			// which serves it.
			var fetcherOptions []fetcher.Option
			opportunityStore, openErr := cli.OpenConfiguredStore(ctx, conf)
			switch {
			case errors.Is(openErr, cli.ErrNoDatabase):
				slog.Warn(
					"opportunities aren't stored or served", "reason", openErr,
				)
			case openErr != nil:
				return openErr
			default:
				defer func() {
					err = errors.Join(err, opportunityStore.Close())
				}()
				fetcherOptions = append(
					fetcherOptions, fetcher.WithStore(opportunityStore),
				)
				options = append(
					options, server.WithStore(opportunityStore),
				)
			}

			// Events from the sources and the API, such as new opportunities and
			// changed marks, are pushed to every WebSocket client.
			var bus events.Bus
			options = append(options, server.WithEvents(&bus))

			appServer, err := server.New(&conf.Server, options...)
			if err != nil {
				return err
			}

			// runCtx stops whichever of the server and the sources is still
			// running once the other has stopped.
			runCtx, cancelRun := context.WithCancel(ctx)
			defer cancelRun()

			fetcherOptions = append(fetcherOptions, fetcher.WithEvents(&bus))

			sourceFetcher := fetcher.New(&conf.Sources, fetcherOptions...)
			fetcherDone := make(chan error, 1)
			go func() {
				fetchErr := sourceFetcher.Run(runCtx, since)
				// Only a failure stops the server: with no sources enabled, Run
				// returns nil straight away.
				if fetchErr != nil {
					cancelRun()
				}
				fetcherDone <- fetchErr
			}()

			serverErr := appServer.Run(runCtx)
			cancelRun()

			return errors.Join(serverErr, <-fetcherDone)
		},
	}

	flags := cmd.Flags()
	flags.String(
		"listen", "", "listen address (default from config, 127.0.0.1:9393)",
	)
	flags.String(
		"host", "", "listen address (default from config, 127.0.0.1)",
	)
	flags.Int("port", 0, "listen address (default from config, 9393)")
	flags.Bool("foreground", false, "run in the foreground")
	flags.String(
		"access-log",
		"",
		`access log file, or "-" for stdout; "" for none `+
			"(default from config, server.access_log)",
	)
	flags.Time(
		"since",
		time.Time{},
		[]string{
			time.DateOnly, time.DateTime,
		},
		"contract updates published since YYYY-MM-DD[ hh:mm:ss] "+
			"(default: each source's newest stored release, or the last day)",
	)
	cmd.MarkFlagsMutuallyExclusive("listen", "host")
	cmd.MarkFlagsMutuallyExclusive("listen", "port")

	return cmd
}

// serverOptions returns the server's options for conf: the access log, if
// server.access_log is set, and the frontend's files, if server.frontend.dir
// is (checked first). closeAccessLog closes the access log; call it once the
// server has stopped.
func serverOptions(
	conf *config.JSCServer,
) ([]server.Option, func() error, error) {
	var options []server.Option
	closeAccessLog := func() error {
		return nil
	}
	if conf.AccessLog != "" {
		accessLog, closeFile, err := logging.OpenAccessLog(conf.AccessLog)
		if err != nil {
			return nil, nil, err
		}
		closeAccessLog = closeFile
		options = append(options, server.WithAccessLog(accessLog))
	}

	if dir := conf.Frontend.Dir; dir != "" {
		files := os.DirFS(dir)
		err := frontend.CheckFS(files)
		if err != nil {
			return nil, nil, errors.Join(
				fmt.Errorf("server.frontend.dir %s: %w", dir, err),
				closeAccessLog(),
			)
		}
		options = append(options, server.WithFrontend(files))
	}

	return options, closeAccessLog, nil
}