pocket-id/backend/internal/bootstrap/bootstrap.go

package bootstrap

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"time"

	_ "github.com/golang-migrate/migrate/v4/source/file"

	"github.com/italypaleale/francis/components"
	francishost "github.com/italypaleale/francis/host"
	"github.com/italypaleale/go-kit/servicerunner"
	"gorm.io/gorm"

	"github.com/pocket-id/pocket-id/backend/internal/common"
	"github.com/pocket-id/pocket-id/backend/internal/instanceid"
	"github.com/pocket-id/pocket-id/backend/internal/storage"
)

func Bootstrap(ctx context.Context) error {
	// List of services to run
	services := make([]servicerunner.Service, 0, 3)
	shutdowns := &shutdownManager{
		fns: make([]servicerunner.Service, 0, 4),
	}

	// Initialize the observability stack, including the logger, distributed tracing, and metrics
	shutdownFns, httpClient, err := initObservability(ctx)
	if err != nil {
		return fmt.Errorf("failed to initialize OpenTelemetry: %w", err)
	}
	shutdowns.Add(shutdownFns...)

	slog.InfoContext(ctx, "Pocket ID is starting")

	// Init database
	db, pg, err := NewDatabase(ctx)
	if err != nil {
		return fmt.Errorf("failed to initialize database: %w", err)
	}
	if pg != nil {
		defer func() {
			// Close the database connection pool only after the shutdown functions have run: some of them (e.g. the actor host deregistering itself from the cluster) still need to query the database.
			pg.Close()
		}()
	}

	// Load the instance ID
	// This is stored in the "kv" table, and generated on first startup
	instanceID, err := instanceid.Load(ctx, db)
	if err != nil {
		return fmt.Errorf("failed to initialize instance ID: %w", err)
	}

	// Init storage
	fileStorage, err := InitStorage(ctx, db)
	if err != nil {
		return fmt.Errorf("failed to initialize file storage (backend: %s): %w", common.EnvConfig.FileBackend, err)
	}

	// Close file storage after every service that depends on it has stopped
	defer func() {
		closeErr := fileStorage.Close()
		if closeErr != nil {
			slog.ErrorContext(ctx, "Failed to close file storage", slog.Any("error", closeErr))
		}
	}()

	// Init application images
	imageExtensions, err := initApplicationImages(ctx, fileStorage)
	if err != nil {
		return fmt.Errorf("failed to initialize application images: %w", err)
	}

	// Init the actors
	// The actor host is created and started before the services, so services can depend on it once it's ready
	actorsOpts := NewActorsOpts{
		Postgres: pg,

		EnvConfig:   &common.EnvConfig,
		InstanceID:  instanceID,
		HttpClient:  httpClient,
		DB:          db,
		FileStorage: fileStorage,
	}
	actors, rateLimitServices, err := NewActors(actorsOpts)
	if err != nil {
		return fmt.Errorf("failed to initialize actors: %w", err)
	}

	// Run the actor host as a background service and get a "ready" signal that other services can wait on
	actorsRun, actorsReady := actorsRunServiceFn(actors)
	services = append(services, actorsRun)

	// Create all services
	svc, err := initServices(ctx, db, instanceID, actors, httpClient, imageExtensions, fileStorage)
	if err != nil {
		return fmt.Errorf("failed to initialize services: %w", err)
	}

	// Migrate the pre-actor signup tokens into their actors, once the actor host is ready
	services = append(services, actorsReady.Await(svc.userSignUpModule.RunSignupTokenMigration))

	// These services are only registered in non-test mode
	if common.EnvConfig.AppEnv != "test" {
		// Refresh the GeoLite database (this is cached per each replica)
		services = append(services, svc.geoLiteModule.Run)
	}

	// Init the router
	// The rate-limit middleware invokes the actor host with each request's own context, so the setup context is intentionally not threaded through the router
	//nolint:contextcheck
	router, err := initRouter(db, svc, rateLimitServices)
	if err != nil {
		return fmt.Errorf("failed to initialize router: %w", err)
	}

	// The router must wait on the actor host being ready, since the rate-limit middleware invokes actors
	services = append(services, actorsReady.Await(router))

	// Run all background services
	// This call blocks until the context is canceled
	err = servicerunner.NewServiceRunner(services...).Run(ctx)
	if errors.Is(err, components.ErrClusterFull) {
		// TODO: Once HA mode is supported, add a note about enabling it
		return errors.New("it appears that there's already one instance of Pocket ID running - running multiple replicas is not (yet) supported")
	} else if err != nil {
		return fmt.Errorf("failed to run services: %w", err)
	}

	// Run all shutdown functions
	shutdowns.Run(ctx)

	return nil
}

// actorsRunServiceFn wraps the actor host's Run method in a background service and returns a "ready" signal that other services can wait on
func actorsRunServiceFn(actors francishost.Host) (servicerunner.Service, *servicerunner.Ready) {
	actorsReady := servicerunner.NewReady()
	fn := func(ctx context.Context) error {
		runErrCh := make(chan error, 1)
		go func() {
			runErrCh <- actors.Run(ctx)
		}()

		// Wait for the right signal
		select {
		case <-actors.Ready():
			// Actor host is ready, signal actorsReady
			actorsReady.Signal()
		case runErr := <-runErrCh:
			// Run returned with an error
			return runErr
		case <-ctx.Done():
			// Context canceled
			return ctx.Err()
		}

		// Now the actor host is running
		// This goroutine must stay up until the actor host returns
		// Here, context cancellation will surface through this channel too
		return <-runErrCh
	}

	return fn, actorsReady
}

func InitStorage(ctx context.Context, db *gorm.DB) (fileStorage storage.FileStorage, err error) {
	switch common.EnvConfig.FileBackend {
	case storage.TypeFileSystem:
		fileStorage, err = storage.NewFilesystemStorage(common.EnvConfig.UploadPath)
	case storage.TypeDatabase:
		fileStorage, err = storage.NewDatabaseStorage(db)
	case storage.TypeS3:
		s3Cfg := storage.S3Config{
			Bucket:                        common.EnvConfig.S3Bucket,
			Region:                        common.EnvConfig.S3Region,
			Endpoint:                      common.EnvConfig.S3Endpoint,
			AccessKeyID:                   common.EnvConfig.S3AccessKeyID,
			SecretAccessKey:               common.EnvConfig.S3SecretAccessKey,
			ForcePathStyle:                common.EnvConfig.S3ForcePathStyle,
			DisableDefaultIntegrityChecks: common.EnvConfig.S3DisableDefaultIntegrityChecks,
			Root:                          common.EnvConfig.UploadPath,
		}
		fileStorage, err = storage.NewS3Storage(ctx, s3Cfg)
	default:
		err = fmt.Errorf("unknown file storage backend: %s", common.EnvConfig.FileBackend)
	}
	if err != nil {
		return fileStorage, err
	}

	return fileStorage, nil
}

type shutdownManager struct {
	fns []servicerunner.Service
}

func (s *shutdownManager) Add(fns ...servicerunner.Service) {
	for _, fn := range fns {
		if fn == nil {
			continue
		}

		s.fns = append(s.fns, fn)
	}
}

func (s *shutdownManager) Run(ctx context.Context) {
	// Cleanup functions are one-shot and must each run to completion independently, so we set WaitAll to true
	sr := servicerunner.NewServiceRunner(s.fns...)
	sr.WaitAll = true

	shutdownCtx, shutdownCancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
	defer shutdownCancel()
	err := sr.Run(shutdownCtx)
	if err != nil {
		slog.ErrorContext(ctx, "Error shutting down services", slog.Any("error", err))
	}
}