app.go

package app

import (
	"context"
	"fmt"
	"log/slog"
	"math"
	"net/http"
	"os"
	"os/signal"
	"strconv"
	"strings"
	"syscall"
	"time"

	"github.com/getkin/kin-openapi/openapi3"
	"github.com/indexdata/crosslink/broker/catalog"
	"github.com/indexdata/crosslink/broker/email"
	prapi "github.com/indexdata/crosslink/broker/patron_request/api"
	pr_db "github.com/indexdata/crosslink/broker/patron_request/db"
	"github.com/indexdata/crosslink/broker/patron_request/proapi"
	prservice "github.com/indexdata/crosslink/broker/patron_request/service"
	psapi "github.com/indexdata/crosslink/broker/pullslip/api"
	ps_db "github.com/indexdata/crosslink/broker/pullslip/db"
	psoapi "github.com/indexdata/crosslink/broker/pullslip/oapi"
	schedapi "github.com/indexdata/crosslink/broker/scheduler/api"
	sched_db "github.com/indexdata/crosslink/broker/scheduler/db"
	schedoapi "github.com/indexdata/crosslink/broker/scheduler/oapi"
	sched_service "github.com/indexdata/crosslink/broker/scheduler/service"
	"github.com/indexdata/crosslink/broker/tenant"

	"github.com/dustin/go-humanize"
	"github.com/indexdata/crosslink/broker/adapter"
	"github.com/indexdata/crosslink/broker/api"
	"github.com/indexdata/crosslink/broker/client"
	"github.com/indexdata/crosslink/broker/dbutil"
	"github.com/indexdata/crosslink/broker/oapi"
	"github.com/indexdata/crosslink/broker/service"
	"github.com/indexdata/crosslink/broker/vcs"

	_ "github.com/golang-migrate/migrate/v4/database/postgres"
	_ "github.com/golang-migrate/migrate/v4/source/file"
	"github.com/indexdata/crosslink/broker/common"
	"github.com/indexdata/crosslink/broker/events"
	"github.com/indexdata/crosslink/broker/handler"
	"github.com/indexdata/crosslink/broker/ill_db"
	"github.com/indexdata/go-utils/utils"
	"github.com/jackc/pgx/v5/pgxpool"
	nethttpmiddleware "github.com/oapi-codegen/nethttp-middleware"

	"github.com/indexdata/crosslink/broker/lms"
)

var HTTP_PORT = utils.Must(utils.GetEnvInt("HTTP_PORT", 8081))
var DB_TYPE = utils.GetEnv("DB_TYPE", "postgres")
var DB_USER = utils.GetEnv("DB_USER", "crosslink")
var METAPROXY_URL = utils.GetEnv("METAPROXY_URL", "")
var DB_PASSWORD = utils.GetEnv("DB_PASSWORD", "crosslink")
var DB_HOST = utils.GetEnv("DB_HOST", "localhost")
var DB_PORT = utils.GetEnv("DB_PORT", "25432")
var DB_DATABASE = utils.GetEnv("DB_DATABASE", "crosslink")
var DB_SCHEMA = utils.GetEnv("DB_SCHEMA", "crosslink_broker")
var DB_EXPLAIN_ANALYZE, _ = utils.GetEnvBool("DB_EXPLAIN_ANALYZE", false)
var ConnectionString = dbutil.GetConnectionString(DB_TYPE, DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_DATABASE, DB_SCHEMA)
var API_PAGE_SIZE int32 = int32(utils.Must(utils.GetEnvInt("API_PAGE_SIZE", int(api.LIMIT_DEFAULT))))
var MigrationsFolder = "file://migrations"
var DB_PROVISION, _ = utils.GetEnvBool("DB_PROVISION", false)
var DB_MIGRATE, _ = utils.GetEnvBool("DB_MIGRATE", true)
var ENABLE_JSON_LOG = utils.GetEnv("ENABLE_JSON_LOG", "false")
var LOG_LEVEL = utils.GetEnv("LOG_LEVEL", "INFO")
var HOLDINGS_ADAPTER = utils.GetEnv("HOLDINGS_ADAPTER", "mock")
var HOLDINGS_SRU_URL = common.GetEnvWithDeprecated("HOLDINGS_SRU_URL", "SRU_URL", "http://localhost:8081/sru")
var HOLDINGS_ISXN_LOOKUP, _ = utils.GetEnvBool("HOLDINGS_ISXN_LOOKUP", false)
var HOLDINGS_FORMAT = utils.GetEnv("HOLDINGS_FORMAT", "reservoir")
var CONSORTIUM_SYMBOL = utils.GetEnv("CONSORTIUM_SYMBOL", "")
var DIRECTORY_ADAPTER = utils.GetEnv("DIRECTORY_ADAPTER", "mock")
var AVAILABILITY_ADAPTER = utils.GetEnv("AVAILABILITY_ADAPTER", "zoom")
var DIRECTORY_API_URL = utils.GetEnv("DIRECTORY_API_URL", "http://localhost:8081/directory/entries")
var MAX_MESSAGE_SIZE, _ = utils.GetEnvAny("MAX_MESSAGE_SIZE", int(100*1024), func(val string) (int, error) {
	v, err := humanize.ParseBytes(val)
	if err != nil && v > uint64(math.MaxInt) {
		appCtx.Logger().Error("MAX_MESSAGE_SIZE value is too large, using default")
		return 0, fmt.Errorf("value %s is too large", val)
	}
	return int(v), err
})
var BROKER_MODE = utils.GetEnv("BROKER_MODE", "opaque")
var TENANT_TO_SYMBOL = os.Getenv("TENANT_TO_SYMBOL")
var CLIENT_DELAY = utils.GetEnv("CLIENT_DELAY", "0ms")
var SHUTDOWN_DELAY, _ = utils.GetEnvAny("SHUTDOWN_DELAY", time.Duration(15*float64(time.Second)), func(val string) (time.Duration, error) {
	d, err := time.ParseDuration(val)
	if err != nil {
		return 0, fmt.Errorf("invalid SHUTDOWN_DELAY value: %s", val)
	}
	return d, nil
})

var ServeMux *http.ServeMux
var appCtx = common.CreateExtCtxWithLogArgsAndHandler(context.Background(), nil, configLog())

type Context struct {
	EventBus        events.EventBus
	IllRepo         ill_db.IllRepo
	EventRepo       events.EventRepo
	DirAdapter      adapter.DirectoryLookupAdapter
	PrRepo          pr_db.PrRepo
	TenantResolver  *tenant.TenantResolver
	ApiHandler      api.ApiHandler
	PrApiHandler    prapi.PatronRequestApiHandler
	SseBroker       *api.SseBroker
	PsApiHandler    psapi.PullSlipApiHandler
	SchedApiHandler schedapi.SchedulerApiHandler
}

func configLog() slog.Handler {
	var level slog.Level
	switch strings.ToUpper(LOG_LEVEL) {
	case "DEBUG":
		level = slog.LevelDebug
	case "INFO":
		level = slog.LevelInfo
	case "WARN":
		level = slog.LevelWarn
	case "ERROR":
		level = slog.LevelError
	default:
		level = slog.LevelInfo
	}
	opts := &slog.HandlerOptions{
		Level: level,
	}
	if strings.EqualFold(ENABLE_JSON_LOG, "true") {
		jsonHandler := slog.NewJSONHandler(os.Stdout, opts)
		common.DefaultLogHandler = jsonHandler
		return jsonHandler
	} else {
		textHandler := slog.NewTextHandler(os.Stdout, opts)
		common.DefaultLogHandler = textHandler
		return textHandler
	}
}

func Init(ctx context.Context) (Context, error) {
	appCtx.Logger().Info("starting " + vcs.GetSignature())
	lookupAdapterEnv, err := catalog.CreateLookupAdapterFromEnv(map[string]any{
		catalog.HoldingsAdapter:    HOLDINGS_ADAPTER,
		catalog.HoldingsSruURL:     HOLDINGS_SRU_URL,
		catalog.HoldingsIsxnLookup: HOLDINGS_ISXN_LOOKUP,
		catalog.HoldingsFormat:     HOLDINGS_FORMAT,
	})
	if err != nil {
		return Context{}, err
	}

	adapter.DEFAULT_BROKER_MODE = getBrokerMode(BROKER_MODE)
	dirAdapter, err := adapter.CreateDirectoryLookupAdapter(map[string]string{
		adapter.DirectoryAdapter: DIRECTORY_ADAPTER,
		adapter.DirectoryApiUrl:  DIRECTORY_API_URL,
	})
	if err != nil {
		return Context{}, err
	}

	delay, err := time.ParseDuration(CLIENT_DELAY)
	if err != nil {
		return Context{}, err
	}

	err = RunDbUp()
	if err != nil {
		return Context{}, err
	}

	pool, err := InitDbPool()
	if err != nil {
		return Context{}, err
	}

	eventRepo := CreateEventRepo(pool)
	eventBus := CreateEventBus(eventRepo)
	illRepo := ill_db.CreateIllRepo(pool)
	prRepo := pr_db.CreatePrRepo(pool, DB_EXPLAIN_ANALYZE)
	psRepo := ps_db.CreatePsRepo(pool)
	schedRepo := sched_db.CreateSchedRepo(pool)

	var emailSenderService *sched_service.EmailSenderService
	emailSenderService, err = sched_service.NewEmailSenderService(prRepo, illRepo)

	prMessageHandler := prservice.CreatePatronRequestMessageHandler(prRepo, eventRepo, illRepo, eventBus)
	iso18626Handler := handler.CreateIso18626Handler(eventBus, eventRepo, illRepo, dirAdapter)
	lmsCreator := lms.NewLmsCreator(illRepo, dirAdapter)
	lookupAdapterCreator := catalog.NewLookupAdapterCreator(AVAILABILITY_ADAPTER, METAPROXY_URL)
	prActionService := prservice.CreatePatronRequestActionService(prRepo, illRepo, eventBus, &iso18626Handler, lmsCreator, email.NewEmailService())
	prMessageHandler.SetAutoActionRunner(prActionService)
	iso18626Client := client.CreateIso18626Client(eventBus, illRepo, prMessageHandler, MAX_MESSAGE_SIZE, delay)
	lookupAdapterFactory := service.NewLookupAdapterFactory(illRepo, dirAdapter, CONSORTIUM_SYMBOL, lookupAdapterEnv, lookupAdapterCreator)
	supplierLocator := service.CreateSupplierLocator(eventBus, illRepo, dirAdapter, lookupAdapterFactory)
	workflowManager := service.CreateWorkflowManager(eventBus, illRepo, service.WorkflowConfig{})
	tenantResolver := tenant.NewResolver().WithIllRepo(illRepo).WithLookupAdapter(dirAdapter).WithTenantToSymbol(TENANT_TO_SYMBOL)
	apiHandler := api.NewApiHandler(eventRepo, illRepo, tenantResolver, API_PAGE_SIZE)
	prApiHandler := prapi.NewPrApiHandler(prRepo, eventBus, eventRepo, tenantResolver, &iso18626Handler, API_PAGE_SIZE)
	prApiHandler.SetAutoActionRunner(prActionService)
	prApiHandler.SetActionTaskProcessor(prActionService)
	prApiHandler.SetIllRepo(illRepo)
	prApiHandler.SetDirectoryLookupAdapter(dirAdapter)
	prApiHandler.SetLookupAdapterFactory(lookupAdapterFactory)
	sseBroker := api.NewSseBroker(appCtx, tenantResolver)
	psApiHandler := psapi.NewPsApiHandler(psRepo, prRepo, tenantResolver)

	batchActionService := sched_service.NewBatchActionService(eventBus, prRepo, schedRepo, emailSenderService)

	if err != nil {
		appCtx.Logger().Warn("email service not available, email sending events will fail", "error", err)
	}

	AddDefaultHandlers(eventBus, iso18626Client, supplierLocator, workflowManager, iso18626Handler, sseBroker, batchActionService, *prActionService)
	err = StartEventBus(ctx, eventBus)
	if err != nil {
		return Context{}, err
	}

	schedApiHandler := schedapi.NewSchedulerApiHandler(API_PAGE_SIZE, schedRepo, eventRepo, tenantResolver)
	if err = StartScheduler(ctx, schedRepo, eventBus); err != nil {
		return Context{}, err
	}

	return Context{
		EventBus:        eventBus,
		IllRepo:         illRepo,
		EventRepo:       eventRepo,
		DirAdapter:      dirAdapter,
		PrRepo:          prRepo,
		TenantResolver:  tenantResolver,
		ApiHandler:      apiHandler,
		PrApiHandler:    prApiHandler,
		SseBroker:       sseBroker,
		PsApiHandler:    psApiHandler,
		SchedApiHandler: schedApiHandler,
	}, nil
}

func Run(ctx context.Context) error {
	context, err := Init(ctx)
	if err != nil {
		return err
	}
	return StartServer(context)
}

func StartServer(ctx Context) error {
	ServeMux = http.NewServeMux()
	oapiValidator, err := newOpenAPIRequestValidator()
	if err != nil {
		return err
	}
	ServeMux.HandleFunc("GET /healthz", HandleHealthz)
	//all methods must be mapped explicitly to avoid conflicts with the index handler
	ServeMux.HandleFunc("GET /iso18626", handler.Iso18626PostHandler(ctx.IllRepo, ctx.EventBus, ctx.DirAdapter, MAX_MESSAGE_SIZE))
	ServeMux.HandleFunc("POST /iso18626", handler.Iso18626PostHandler(ctx.IllRepo, ctx.EventBus, ctx.DirAdapter, MAX_MESSAGE_SIZE))
	ServeMux.HandleFunc("PUT /iso18626", handler.Iso18626PostHandler(ctx.IllRepo, ctx.EventBus, ctx.DirAdapter, MAX_MESSAGE_SIZE))
	ServeMux.HandleFunc("DELETE /iso18626", handler.Iso18626PostHandler(ctx.IllRepo, ctx.EventBus, ctx.DirAdapter, MAX_MESSAGE_SIZE))
	ServeMux.HandleFunc("GET /v3/open-api.yaml", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/x-yaml")
		_, _ = w.Write(oapi.OpenAPISpecYAML)
	})
	oapi.HandlerFromMux(&ctx.ApiHandler, ServeMux)
	proapi.HandlerWithOptions(&ctx.PrApiHandler, proapi.StdHTTPServerOptions{
		BaseRouter:  ServeMux,
		Middlewares: []proapi.MiddlewareFunc{oapiValidator},
	})
	psoapi.HandlerFromMux(&ctx.PsApiHandler, ServeMux)
	schedoapi.HandlerFromMux(&ctx.SchedApiHandler, ServeMux)
	ServeMux.HandleFunc("GET /sse/events", ctx.SseBroker.ServeHTTP)
	if ctx.TenantResolver.HasTenantMapping() {
		basePath := tenant.OKAPI_PATH_PREFIX
		oapi.HandlerFromMuxWithBaseURL(&ctx.ApiHandler, ServeMux, basePath)
		proapi.HandlerWithOptions(&ctx.PrApiHandler, proapi.StdHTTPServerOptions{
			BaseURL:     basePath,
			BaseRouter:  ServeMux,
			Middlewares: []proapi.MiddlewareFunc{oapiValidator},
		})
		psoapi.HandlerFromMuxWithBaseURL(&ctx.PsApiHandler, ServeMux, basePath)
		schedoapi.HandlerFromMuxWithBaseURL(&ctx.SchedApiHandler, ServeMux, basePath)
		ServeMux.HandleFunc("GET "+basePath+"/sse/events", ctx.SseBroker.ServeHTTP)
	}
	signatureHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Server", vcs.GetSignature())
		ServeMux.ServeHTTP(w, r)
	})
	server := &http.Server{
		Addr:              ":" + strconv.Itoa(HTTP_PORT),
		Handler:           signatureHandler,
		ReadHeaderTimeout: 10 * time.Second,
		ReadTimeout:       30 * time.Second,
		WriteTimeout:      60 * time.Second,
		IdleTimeout:       120 * time.Second,
	}
	// channel to listen for server errors
	serverErrors := make(chan error, 1)
	go func() {
		appCtx.Logger().Info("HTTP server started on port " + strconv.Itoa(HTTP_PORT))
		serverErrors <- server.ListenAndServe()
	}()
	// channel to listen for OS signals
	shutdown := make(chan os.Signal, 1)
	signal.Notify(shutdown, os.Interrupt, syscall.SIGTERM)
	// block until we receive a signal or server error
	select {
	case err := <-serverErrors:
		return fmt.Errorf("HTTP server error: %w", err)
	case sig := <-shutdown:
		appCtx.Logger().Info("HTTP server shutdown initiated", "signal", sig)
		// give outstanding requests a timeout to complete
		ctx, cancel := context.WithTimeout(appCtx, SHUTDOWN_DELAY)
		defer cancel()
		if err := server.Shutdown(ctx); err != nil {
			server.Close()
			return fmt.Errorf("HTTP server could not shutdown gracefully: %w", err)
		}
		appCtx.Logger().Info("HTTP server shutdown complete")
		return nil
	}
}

func newOpenAPIRequestValidator() (func(http.Handler) http.Handler, error) {
	spec, err := openapi3.NewLoader().LoadFromData(oapi.OpenAPISpecYAML)
	if err != nil {
		return nil, err
	}
	spec.Servers = openapi3.Servers{
		{URL: "/"},
		{URL: tenant.OKAPI_PATH_PREFIX},
	}
	return nethttpmiddleware.OapiRequestValidatorWithOptions(spec, &nethttpmiddleware.Options{
		ErrorHandlerWithOpts:  writeOpenAPIValidationError,
		SilenceServersWarning: true,
	}), nil
}

func writeOpenAPIValidationError(_ context.Context, err error, w http.ResponseWriter, _ *http.Request, opts nethttpmiddleware.ErrorHandlerOpts) {
	statusCode := opts.StatusCode
	if statusCode == 0 {
		statusCode = http.StatusBadRequest
	}
	api.WriteJsonErrorResponse(w, err, statusCode)
}

func RunDbUp() error {
	if !DB_PROVISION && !DB_MIGRATE {
		appCtx.Logger().Info("DB up skipped", "dbProvision", DB_PROVISION, "dbMigrate", DB_MIGRATE)
		return nil
	}
	if DB_PROVISION {
		err := dbutil.RunDbProvision(ConnectionString, DB_SCHEMA)
		if err != nil {
			return fmt.Errorf("DB provision failed: err=%w", err)
		}
		appCtx.Logger().Info("DB provision success", "dbProvision", DB_PROVISION)
	}
	if !DB_MIGRATE {
		appCtx.Logger().Info("DB migration skipped", "dbMigrate", DB_MIGRATE)
		return nil
	}
	verFrom, verTo, dirty, err := dbutil.RunDbMigrations(MigrationsFolder, ConnectionString)
	if err != nil {
		return fmt.Errorf("DB migration failed: err=%w versionFrom=%d versionTo=%d dirty=%t", err, verFrom, verTo, dirty)
	}
	appCtx.Logger().Info("DB migration success", "versionFrom", verFrom, "versionTo", verTo, "dirty", dirty)
	return nil
}

func InitDbPool() (*pgxpool.Pool, error) {
	dbPool, err := dbutil.InitDbPool(ConnectionString)
	if err != nil {
		return nil, fmt.Errorf("unable to create pool to database: %w", err)
	}
	return dbPool, nil
}

func CreateEventRepo(dbPool *pgxpool.Pool) events.EventRepo {
	eventRepo := new(events.PgEventRepo)
	eventRepo.Pool = dbPool
	return eventRepo
}

func CreateEventBus(eventRepo events.EventRepo) events.EventBus {
	eventBus := events.NewPostgresEventBus(eventRepo, ConnectionString)
	return eventBus
}

func AddDefaultHandlers(eventBus events.EventBus, iso18626Client client.Iso18626Client,
	supplierLocator service.SupplierLocator, workflowManager service.WorkflowManager, iso18626Handler handler.Iso18626Handler,
	sseBroker *api.SseBroker, batchActionService *sched_service.BatchActionService, prActionService prservice.PatronRequestActionService) {
	eventBus.HandleEventCreated(events.EventNameMessageSupplier, events.HandlerRoleConsumer, iso18626Client.MessageSupplier)
	eventBus.HandleEventCreated(events.EventNameMessageRequester, events.HandlerRoleConsumer, iso18626Client.MessageRequester)
	eventBus.HandleEventCreated(events.EventNameConfirmRequesterMsg, events.HandlerRoleObserver, iso18626Handler.ConfirmRequesterMsg)
	eventBus.HandleEventCreated(events.EventNameConfirmSupplierMsg, events.HandlerRoleObserver, iso18626Handler.ConfirmSupplierMsg)

	eventBus.HandleEventCreated(events.EventNameLocateSuppliers, events.HandlerRoleConsumer, supplierLocator.LocateSuppliers)
	eventBus.HandleEventCreated(events.EventNameSelectSupplier, events.HandlerRoleConsumer, supplierLocator.SelectSupplier)
	eventBus.HandleEventCreated(events.EventNameCheckAvailability, events.HandlerRoleConsumer, supplierLocator.CheckAvailability)

	eventBus.HandleEventCreated(events.EventNameRequestReceived, events.HandlerRoleConsumer, workflowManager.RequestReceived)
	eventBus.HandleEventCreated(events.EventNameSupplierMsgReceived, events.HandlerRoleConsumer, workflowManager.SupplierMessageReceived)
	eventBus.HandleEventCreated(events.EventNameRequesterMsgReceived, events.HandlerRoleConsumer, workflowManager.RequesterMessageReceived)
	eventBus.HandleTaskCompleted(events.EventNameLocateSuppliers, events.HandlerRoleConsumer, workflowManager.OnLocateSupplierComplete)
	eventBus.HandleTaskCompleted(events.EventNameSelectSupplier, events.HandlerRoleConsumer, workflowManager.OnSelectSupplierComplete)
	eventBus.HandleTaskCompleted(events.EventNameCheckAvailability, events.HandlerRoleConsumer, workflowManager.OnCheckAvailabilityComplete)
	eventBus.HandleTaskCompleted(events.EventNameMessageSupplier, events.HandlerRoleConsumer, workflowManager.OnMessageSupplierComplete)
	eventBus.HandleTaskCompleted(events.EventNameMessageRequester, events.HandlerRoleConsumer, workflowManager.OnMessageRequesterComplete)
	eventBus.HandleTaskCompleted(events.EventNameMessageSupplier, events.HandlerRoleObserver, sseBroker.IncomingIsoMessage)
	eventBus.HandleTaskCompleted(events.EventNameMessageRequester, events.HandlerRoleObserver, sseBroker.IncomingIsoMessage)

	eventBus.HandleEventCreated(events.EventNameInvokeBatchAction, events.HandlerRoleConsumer, batchActionService.BatchAction)

	eventBus.HandleEventCreated(events.EventNameInvokeBackgroundAction, events.HandlerRoleConsumer, prActionService.InvokeAction)

	// Invoke-action is intentionally not registered on event-created/task-completed handlers.
	// It is processed inline by patron-request services and API handlers.
}
func StartEventBus(ctx context.Context, eventBus events.EventBus) error {
	err := eventBus.Start(common.CreateExtCtxWithArgs(ctx, nil))
	if err != nil {
		return fmt.Errorf("starting event bus failed err=%w", err)
	}
	return nil
}

// StartScheduler creates the scheduler service, begins listening on
// sched_db.SchedulerChannel, and launches the scheduling loop in a background goroutine.
func StartScheduler(ctx context.Context, schedRepo sched_db.SchedRepo, eventBus events.EventBus) error {
	extCtx := common.CreateExtCtxWithArgs(ctx, nil)
	svc := sched_service.NewSchedulerService(schedRepo, eventBus, ConnectionString)
	if err := svc.Listen(extCtx); err != nil {
		return fmt.Errorf("starting scheduler listener failed: %w", err)
	}
	go svc.Run(extCtx)
	return nil
}

func HandleHealthz(w http.ResponseWriter, r *http.Request) {
	w.Write([]byte("OK"))
}

func getBrokerMode(mode string) common.BrokerMode {
	if strings.EqualFold(mode, string(common.BrokerModeTransparent)) {
		return common.BrokerModeTransparent
	} else {
		return common.BrokerModeOpaque
	}
}