// Package integrationservice assembles the independently deployable,
// provider-facing Integration role. It owns webhook protocol and external
// connector effects; its only business-operation handoff to L0 is the private
// neutral Platform Runtime client.
package integrationservice

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"net/http"
	"sync"
	"sync/atomic"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/google/uuid"
	"github.com/nexuscrm/backend/internal/runtimeprocess"
	database "github.com/nexuscrm/db"
	httpclient "github.com/nexuscrm/outbound/http_client"
	"github.com/nexuscrm/outbound/url_safety"
	"github.com/pingcap-inc/nexuscrm-api-host/processhttp"
	"github.com/pingcap-inc/nexuscrm-api-host/runtimegrpc"
	appconnectors "github.com/pingcap-inc/nexuscrm-connectors/application/connectors"
	connectorcallout "github.com/pingcap-inc/nexuscrm-connectors/features/connector/callout"
	connectorcapabilities "github.com/pingcap-inc/nexuscrm-connectors/features/connector/capabilities"
	connectorevents "github.com/pingcap-inc/nexuscrm-connectors/features/connector/events"
	handlerconfig "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundhandlerconfig"
	connectorrouting "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"
	connectorreceipts "github.com/pingcap-inc/nexuscrm-connectors/features/connector/ingressreceipts"
	connectoroutboundintent "github.com/pingcap-inc/nexuscrm-connectors/features/connector/outboundintent"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/packstate"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/runtimehandoff"
	connectorthrottle "github.com/pingcap-inc/nexuscrm-connectors/features/connector/throttle"
	"github.com/pingcap-inc/nexuscrm-connectors/features/namedcredential"
	connectorwebhooks "github.com/pingcap-inc/nexuscrm-connectors/interfaces/rest/connector_webhooks"
	catalogv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	"github.com/prometheus/client_golang/prometheus/promhttp"
)

const Role = "integration-service"

type supervisor interface {
	StartContext(context.Context) error
	Stop() error
}

type receiptRouter interface {
	// Run reports started only after the router has completed an operational
	// probe and entered its long-lived execution path. Returning before that
	// callback is a startup failure, not a running router.
	Run(context.Context, func()) error
}

type receiptRouterLane interface {
	ProcessBatch(context.Context) (int, error)
	Run(context.Context) error
}

type productionReceiptRouter struct {
	router   receiptRouterLane
	outbound interface {
		Run(context.Context, func()) error
	}
}

func (r *productionReceiptRouter) Run(ctx context.Context, started func()) error {
	if r == nil || r.router == nil || r.outbound == nil || started == nil {
		return errors.New("integration service: receipt router startup is unavailable")
	}
	laneCtx, cancel := context.WithCancel(ctx)
	defer cancel()
	type laneResult struct {
		name string
		err  error
	}
	results := make(chan laneResult, 2)
	routerStarted := make(chan struct{})
	outboundStarted := make(chan struct{})
	go func() {
		if _, err := r.router.ProcessBatch(laneCtx); err != nil {
			results <- laneResult{name: "receipt router", err: err}
			return
		}
		close(routerStarted)
		results <- laneResult{name: "receipt router", err: r.router.Run(laneCtx)}
	}()
	go func() {
		err := r.outbound.Run(laneCtx, func() { close(outboundStarted) })
		results <- laneResult{name: "outbound intent owner", err: err}
	}()
	waitStarted := func(ready <-chan struct{}) error {
		select {
		case <-ready:
			return nil
		case result := <-results:
			cancel()
			<-results
			if result.err == nil {
				return fmt.Errorf("integration service: %s stopped during startup", result.name)
			}
			return fmt.Errorf("integration service: %s stopped during startup: %w", result.name, result.err)
		case <-ctx.Done():
			cancel()
			<-results
			<-results
			return ctx.Err()
		}
	}
	if err := waitStarted(routerStarted); err != nil {
		return err
	}
	if err := waitStarted(outboundStarted); err != nil {
		return err
	}
	started()
	first := <-results
	cancel()
	second := <-results
	if ctxErr := ctx.Err(); ctxErr != nil {
		return ctxErr
	}
	for _, result := range []laneResult{first, second} {
		if result.err != nil && !errors.Is(result.err, context.Canceled) {
			return fmt.Errorf("integration service: %s stopped: %w", result.name, result.err)
		}
	}
	return fmt.Errorf("integration service: %s stopped unexpectedly", first.name)
}

type platformClient interface {
	runtimehandoff.PlatformClient
	connectoroutboundintent.IntentSource
	connectoroutboundintent.ResultCommitter
	runtimeConnection
}

type runtimeConnection interface {
	Ready() bool
	Close() error
}

// pinnedPlatformClient keeps command handoff on the exact immutable Catalog
// view that passed process startup verification. A stopped upgrade selects a
// new revision by restarting the process; an individual handoff must never join
// startup authority with a later view.
type pinnedPlatformClient struct {
	runtimehandoff.PlatformClient
	view catalogv1.IntegrationBindingView
}

func newPinnedPlatformClient(client runtimehandoff.PlatformClient, view catalogv1.IntegrationBindingView) *pinnedPlatformClient {
	return &pinnedPlatformClient{PlatformClient: client, view: cloneIntegrationBindingView(view)}
}

func (c *pinnedPlatformClient) GetIntegrationBindingView(_ context.Context, request catalogv1.GetRequest) (catalogv1.IntegrationBindingView, error) {
	if c == nil || c.PlatformClient == nil {
		return catalogv1.IntegrationBindingView{}, errors.New("integration service: pinned Platform authority is unavailable")
	}
	if request.MinCatalogRevision != nil && c.view.Lineage.CatalogRevision < *request.MinCatalogRevision {
		return catalogv1.IntegrationBindingView{}, errors.New("integration service: pinned Platform authority is older than the requested Catalog revision")
	}
	return cloneIntegrationBindingView(c.view), nil
}

func cloneIntegrationBindingView(view catalogv1.IntegrationBindingView) catalogv1.IntegrationBindingView {
	cloned := view
	cloned.Bindings = make([]catalogv1.IntegrationBinding, len(view.Bindings))
	for index := range view.Bindings {
		cloned.Bindings[index] = view.Bindings[index]
		cloned.Bindings[index].CredentialRefs = append([]string(nil), view.Bindings[index].CredentialRefs...)
		if view.Bindings[index].RuntimeCommand != nil {
			command := *view.Bindings[index].RuntimeCommand
			cloned.Bindings[index].RuntimeCommand = &command
		}
	}
	return cloned
}

type httpProcess interface {
	Start() error
	Ready() bool
	Address() string
	Wait() error
	Shutdown(context.Context) error
}

type databaseOwner interface {
	Close() error
}

type Process struct {
	http       httpProcess
	supervisor supervisor
	router     receiptRouter
	runtime    runtimeConnection
	database   databaseOwner

	ready          atomic.Bool
	mu             sync.Mutex
	startAttempted bool
	started        bool
	routerCancel   context.CancelFunc
	routerDone     chan struct{}
	routerRunning  bool
	routerErr      error
	httpDone       chan struct{}
	httpErr        error
	stopOnce       sync.Once
	stopErr        error
}

func AssembleProduction(ctx context.Context, version string, config Config) (*Process, error) {
	if ctx == nil {
		return nil, errors.New("integration service: startup context is required")
	}
	startupCtx, cancel := context.WithTimeout(ctx, config.RuntimeStartupTimeout)
	defer cancel()
	runtimeClient, err := runtimegrpc.DialPlatformRuntime(startupCtx, config.Runtime)
	if err != nil {
		return nil, err
	}
	connection, err := database.NewConnectionContext(startupCtx, productionDatabasePool())
	if err != nil {
		_ = runtimeClient.Close()
		return nil, err
	}
	process, err := assembleProductionWithDatabase(startupCtx, version, config, connection.DB(), runtimeClient, connection)
	if err != nil {
		_ = connection.Close()
		_ = runtimeClient.Close()
		return nil, err
	}
	return process, nil
}

func productionDatabasePool() database.PoolOptions {
	options := database.DefaultPoolOptions()
	// Bound silent driver reads, including COMMIT/ROLLBACK, so durable queue
	// leases and retries can recover without waiting for a remote disconnect.
	options.NetworkTimeout = 30 * time.Second
	return options
}

func assembleProductionWithDatabase(
	ctx context.Context,
	version string,
	config Config,
	db *sql.DB,
	runtimeClient platformClient,
	database databaseOwner,
) (*Process, error) {
	if db == nil || runtimeClient == nil || database == nil {
		return nil, errors.New("integration service: database and Platform Runtime client are required")
	}
	if err := connectorevents.RequireWorkItemOnlyExecutionCutover(ctx, db); err != nil {
		return nil, fmt.Errorf("integration service: WorkItem cutover is not ready: %w", err)
	}
	if err := connectorevents.RequireSourceProjectionProtocol(ctx, db); err != nil {
		return nil, fmt.Errorf("integration service: projection range cutover is not ready: %w", err)
	}
	view, err := readPlatformAuthority(ctx, runtimeClient, config)
	if err != nil {
		return nil, err
	}
	authority, err := newIntegrationServingAuthority(view)
	if err != nil {
		return nil, fmt.Errorf("integration service: build serving authority: %w", err)
	}
	if err := authority.VerifyDatabase(ctx, db); err != nil {
		return nil, fmt.Errorf("integration service: verify serving authority: %w", err)
	}

	cipher, err := namedcredential.NewSecretCipher()
	if err != nil || cipher == nil {
		return nil, fmt.Errorf("integration service: named-credential cipher is required: %w", err)
	}
	registry := connectorcapabilities.NewRegistry(
		authority.CapabilityGate(connectorcapabilities.ProductionGate(db)),
	)
	for _, pack := range authority.servingPacks() {
		if err := handlerconfig.RegisterActiveHandlerConfigsForPackInstall(
			ctx,
			db,
			registry,
			pack.ownerPack,
			pack.packInstallation,
		); err != nil {
			return nil, fmt.Errorf("integration service: register authorized webhook handlers for %s: %w", pack.ownerPack, err)
		}
	}

	events := connectorevents.NewService(db, nil, nil)
	selection, err := authority.RuntimeServingSelection(connectorrouting.NewDBRouteResolver(db))
	if err != nil {
		return nil, fmt.Errorf("integration service: bind route authority: %w", err)
	}
	dispatchPacks := make([]connectorwebhooks.RuntimeServingPackIdentity, 0, len(selection.Packs))
	for _, pack := range selection.Packs {
		dispatchPacks = append(dispatchPacks, connectorwebhooks.RuntimeServingPackIdentity{
			Namespace:       pack.Namespace,
			InstalledPackID: pack.InstalledPackID,
			PayloadDigest:   pack.PayloadDigest,
		})
	}
	dispatcher, err := connectorwebhooks.NewRuntimeServingDispatcher(
		db,
		cipher,
		registry,
		events,
		selection.RouteResolver,
		dispatchPacks,
	)
	if err != nil {
		return nil, fmt.Errorf("integration service: bind webhook pack authority: %w", err)
	}
	claimPackInstallIDs := make([]string, 0, len(selection.Packs))
	for _, pack := range selection.Packs {
		claimPackInstallIDs = append(claimPackInstallIDs, pack.InstalledPackID)
	}
	ingressStore, err := connectorreceipts.NewScopedClaimService(db, cipher, claimPackInstallIDs)
	if err != nil {
		return nil, fmt.Errorf("integration service: bind receipt claim authority: %w", err)
	}
	router, err := connectorwebhooks.NewReceiptRouter(connectorwebhooks.ReceiptRouterConfig{}, ingressStore, dispatcher)
	if err != nil {
		return nil, err
	}

	packResolver := namedcredential.NewControlplanePackResolver(db)
	credentialRepository := namedcredential.NewRepository(db, cipher)
	credentials := namedcredential.NewServiceWithPacks(credentialRepository, nil, packResolver)
	throttler := connectorthrottle.NewService(connectorthrottle.Config{
		Credentials: connectorthrottle.NewNamedCredentialResolver(credentials),
		Limits:      connectorthrottle.NewSQLLimitProvider(db),
	})
	allowlist := url_safety.NewAllowlist()
	callouts := connectorcallout.NewEngine(credentials, throttler, httpclient.NewOutboundClient(allowlist), allowlist)
	outboundScope := connectoroutboundintent.Scope{
		OrgID: config.OrgID, AppInstallationID: config.AppInstallationID,
		AuthorizedWorkloadID: config.AuthorizedWorkloadID,
		CatalogRevision:      view.Lineage.CatalogRevision,
	}
	outboundRepository := connectoroutboundintent.NewRepository(db)
	outboundAuthority, err := connectoroutboundintent.NewAuthority(view)
	if err != nil {
		return nil, fmt.Errorf("integration service: bind outbound intent authority: %w", err)
	}
	providerPreparer, err := connectoroutboundintent.NewProviderPreparer(db, outboundAuthority, callouts)
	if err != nil {
		return nil, fmt.Errorf("integration service: build outbound intent provider owner: %w", err)
	}
	var outboundPreparer connectoroutboundintent.Preparer = providerPreparer
	var mailHandler *connectedMailHandler
	var mailRuntime *connectedMailRuntime
	var attachmentReaders []connectoroutboundintent.MailAttachmentReader
	if config.ConnectedMail != nil {
		attachmentReader, ok := runtimeClient.(connectoroutboundintent.MailAttachmentReader)
		if !ok {
			return nil, errors.New("integration service: Runtime mail attachment capability is unavailable")
		}
		attachmentReaders = append(attachmentReaders, attachmentReader)
	}
	outboundPreparer, mailRuntime, err = withConnectedMailPreparer(
		db, config.ConnectedMail, view, throttler, allowlist, outboundPreparer, attachmentReaders...,
	)
	if err != nil {
		return nil, err
	}
	if mailRuntime != nil {
		mailboxRuntime, ok := runtimeClient.(connectedMailboxAuthorizationRuntime)
		if !ok {
			return nil, errors.New("integration service: Runtime mailbox authorization capability is unavailable")
		}
		state, err := newMailboxStateCodec(config.ConnectedMail.MasterKey)
		if err != nil {
			return nil, fmt.Errorf("integration service: build connected-mail callback state: %w", err)
		}
		mailHandler, err = newConnectedMailHandler(
			config.ConnectedMail,
			mailRuntime.connectorKey,
			view.Lineage.CatalogRevision,
			mailboxRuntime,
			mailRuntime.oauth,
			state,
		)
		if err != nil {
			return nil, err
		}
	}
	outboundAcceptor, err := connectoroutboundintent.NewAcceptor(runtimeClient, outboundRepository, outboundScope)
	if err != nil {
		return nil, fmt.Errorf("integration service: build outbound intent handoff: %w", err)
	}
	outboundWorker, err := connectoroutboundintent.NewWorker(
		outboundRepository, outboundPreparer, runtimeClient, outboundScope,
		Role+":"+uuid.NewString(),
	)
	if err != nil {
		return nil, fmt.Errorf("integration service: build outbound intent worker: %w", err)
	}
	outboundService, err := connectoroutboundintent.NewService(outboundAcceptor, outboundWorker)
	if err != nil {
		return nil, fmt.Errorf("integration service: build outbound intent lifecycle: %w", err)
	}
	handoffStore, err := runtimehandoff.NewStore(db)
	if err != nil {
		return nil, err
	}
	handoff, err := runtimehandoff.NewHandoff(newPinnedPlatformClient(runtimeClient, view), handoffStore, runtimehandoff.Config{
		OrgID: config.OrgID, AppInstallationID: config.AppInstallationID,
		AuthorizedWorkloadID: config.AuthorizedWorkloadID,
	})
	if err != nil {
		return nil, err
	}
	if err := handoff.VerifyReady(ctx); err != nil {
		return nil, fmt.Errorf("integration service: Runtime handoff is not ready: %w", err)
	}
	workers, err := appconnectors.NewRuntimeSupervisor(db, callouts, handoff)
	if err != nil {
		return nil, err
	}
	if err := workers.SetRuntimeServingSelection(selection); err != nil {
		return nil, fmt.Errorf("integration service: bind worker serving authority: %w", err)
	}
	workers.SetBatchAuthTypeResolver(func(ctx context.Context, _ packstate.ConnectorInboundWorkerPackRow, credentialRef string) (string, error) {
		return credentials.ResolveAuthType(ctx, credentialRef)
	})
	workers.SetDrainOnlyEligibilityResolver(packResolver.ResolveDrainOnlyConnectorPackByInstall)

	engine := gin.New()
	engine.Use(gin.Recovery())
	dispatcher.RegisterRoutes(engine.Group("/webhooks"))
	if mailHandler != nil {
		mailHandler.RegisterRoutes(engine)
	}
	return newProcess(
		version, config, engine, workers,
		&productionReceiptRouter{router: router, outbound: outboundService},
		runtimeClient, database,
	)
}

func readPlatformAuthority(ctx context.Context, runtimeClient runtimehandoff.PlatformClient, config Config) (catalogv1.IntegrationBindingView, error) {
	view, err := runtimeClient.GetIntegrationBindingView(ctx, catalogv1.GetRequest{})
	if err != nil {
		return catalogv1.IntegrationBindingView{}, fmt.Errorf("integration service: read Integration binding authority: %w", err)
	}
	if err := catalogv1.ValidateIntegrationBindingView(view); err != nil {
		return catalogv1.IntegrationBindingView{}, fmt.Errorf("integration service: invalid Integration binding authority: %w", err)
	}
	if view.OrgID != config.OrgID || view.AppInstallationID != config.AppInstallationID ||
		view.AuthorizedWorkloadID != config.AuthorizedWorkloadID {
		return catalogv1.IntegrationBindingView{}, errors.New("integration service: Integration binding authority is outside configured consumer scope")
	}
	return view, nil
}

func newProcess(
	version string,
	config Config,
	handler http.Handler,
	workers supervisor,
	router receiptRouter,
	runtimeClient runtimeConnection,
	database databaseOwner,
) (*Process, error) {
	if handler == nil || workers == nil || router == nil || runtimeClient == nil || database == nil {
		return nil, errors.New("integration service: complete process composition is required")
	}
	process := &Process{
		supervisor: workers, router: router, runtime: runtimeClient, database: database,
		routerDone: make(chan struct{}), httpDone: make(chan struct{}),
	}
	// Integration owns connector ingress, receipt routing, inbound execution,
	// provider callouts, and outbound intent processing in this process. Those
	// components register bounded Prometheus collectors on the default registry.
	// Expose them only when the operational listener itself is confined to
	// loopback; a provider-facing listener must not acquire an unauthenticated
	// diagnostics surface merely because it uses the same process lifecycle.
	productHandler := http.NewServeMux()
	metricsHandler := http.NotFoundHandler()
	if runtimeprocess.IsLoopbackListenAddress(config.HTTPAddress) {
		metricsHandler = promhttp.Handler()
	}
	productHandler.Handle("/metrics", metricsHandler)
	productHandler.Handle("/", handler)
	var err error
	process.http, err = processhttp.NewProcess(processhttp.ProcessConfig{
		Role: Role, Version: version, Address: config.HTTPAddress, Handler: productHandler,
		Ready: process.isReady,
		Policy: processhttp.ServerPolicy{
			ReadHeaderTimeout: config.ReadHeaderTimeout, ReadTimeout: config.RequestTimeout,
			WriteTimeout: config.RequestTimeout, IdleTimeout: config.IdleTimeout,
		},
	})
	if err != nil {
		return nil, err
	}
	return process, nil
}

func (p *Process) isReady() bool {
	return p != nil && p.ready.Load() && p.runtime != nil && p.runtime.Ready()
}

func (p *Process) Start(ctx context.Context) error {
	if p == nil || p.http == nil || p.supervisor == nil || p.router == nil || p.runtime == nil {
		return errors.New("integration service: process is unavailable")
	}
	if ctx == nil {
		return errors.New("integration service: startup context is required")
	}
	if err := ctx.Err(); err != nil {
		return err
	}
	p.mu.Lock()
	if p.startAttempted {
		p.mu.Unlock()
		return errors.New("integration service: process has already started")
	}
	p.startAttempted = true
	p.mu.Unlock()
	if !p.runtime.Ready() {
		return errors.New("integration service: Platform Runtime is not ready")
	}
	if err := p.supervisor.StartContext(ctx); err != nil {
		return err
	}
	if err := ctx.Err(); err != nil {
		_ = p.supervisor.Stop()
		return err
	}
	routerCtx, cancel := context.WithCancel(context.Background())
	p.routerCancel = cancel
	routerStarted := make(chan struct{})
	var routerStartedOnce sync.Once
	go func() {
		err := p.router.Run(routerCtx, func() {
			routerStartedOnce.Do(func() {
				p.mu.Lock()
				p.routerRunning = true
				p.mu.Unlock()
				close(routerStarted)
			})
		})
		p.mu.Lock()
		p.routerRunning = false
		p.routerErr = err
		p.ready.Store(false)
		p.mu.Unlock()
		close(p.routerDone)
	}()
	select {
	case <-routerStarted:
	case <-p.routerDone:
		_ = p.supervisor.Stop()
		p.mu.Lock()
		err := p.routerErr
		p.mu.Unlock()
		return receiptRouterStartupExit(err)
	case <-ctx.Done():
		cancel()
		<-p.routerDone
		_ = p.supervisor.Stop()
		return ctx.Err()
	}
	if err := p.http.Start(); err != nil {
		cancel()
		<-p.routerDone
		_ = p.supervisor.Stop()
		return err
	}
	go func() {
		err := p.http.Wait()
		p.mu.Lock()
		p.httpErr = err
		p.ready.Store(false)
		close(p.httpDone)
		p.mu.Unlock()
	}()
	p.mu.Lock()
	routerRunning := p.routerRunning
	routerErr := p.routerErr
	startupErr := ctx.Err()
	httpExited := false
	select {
	case <-p.httpDone:
		httpExited = true
	default:
	}
	runtimeReady := p.runtime.Ready()
	httpReady := false
	if routerRunning && !httpExited && runtimeReady && startupErr == nil {
		// processhttp.Ready also proves that the listener's Serve goroutine is
		// still live. Set the semantic input first, then publish started only if
		// the complete transport + Runtime readiness predicate holds.
		p.ready.Store(true)
		httpReady = p.http.Ready()
		if !httpReady {
			p.ready.Store(false)
		} else {
			p.started = true
			p.mu.Unlock()
			return nil
		}
	}
	p.mu.Unlock()
	cancel()
	_ = p.http.Shutdown(ctx)
	<-p.httpDone
	<-p.routerDone
	_ = p.supervisor.Stop()
	p.mu.Lock()
	httpErr := p.httpErr
	p.mu.Unlock()
	if startupErr != nil {
		return startupErr
	}
	if !runtimeReady {
		return errors.New("integration service: Platform Runtime became unavailable during startup")
	}
	if routerRunning && (httpExited || !httpReady) {
		return httpStartupExit(httpErr)
	}
	return receiptRouterStartupExit(routerErr)
}

func (p *Process) Address() string {
	if p == nil || p.http == nil {
		return ""
	}
	return p.http.Address()
}

func (p *Process) Wait() error {
	if p == nil {
		return errors.New("integration service: process is unavailable")
	}
	p.mu.Lock()
	started := p.started
	p.mu.Unlock()
	if !started {
		return errors.New("integration service: process has not started")
	}
	select {
	case <-p.httpDone:
		p.mu.Lock()
		err := p.httpErr
		p.mu.Unlock()
		return err
	case <-p.routerDone:
		p.mu.Lock()
		err := p.routerErr
		p.mu.Unlock()
		return unexpectedReceiptRouterExit(err)
	}
}

func unexpectedReceiptRouterExit(err error) error {
	if errors.Is(err, context.Canceled) {
		return nil
	}
	if err == nil {
		return errors.New("integration service: receipt router stopped unexpectedly")
	}
	return fmt.Errorf("integration service: receipt router stopped: %w", err)
}

func receiptRouterStartupExit(err error) error {
	if err == nil {
		return errors.New("integration service: receipt router stopped during startup")
	}
	return fmt.Errorf("integration service: receipt router stopped during startup: %w", err)
}

func httpStartupExit(err error) error {
	if err == nil {
		return errors.New("integration service: HTTP server stopped during startup")
	}
	return fmt.Errorf("integration service: HTTP server stopped during startup: %w", err)
}

func (p *Process) Shutdown(ctx context.Context) error {
	if p == nil {
		return nil
	}
	if ctx == nil {
		return errors.New("integration service: shutdown context is required")
	}
	p.stopOnce.Do(func() {
		p.ready.Store(false)
		if p.http != nil {
			p.stopErr = errors.Join(p.stopErr, p.http.Shutdown(ctx))
		}
		if p.routerCancel != nil {
			p.routerCancel()
			select {
			case <-p.routerDone:
				p.mu.Lock()
				err := p.routerErr
				p.mu.Unlock()
				if !errors.Is(err, context.Canceled) {
					p.stopErr = errors.Join(p.stopErr, err)
				}
			case <-ctx.Done():
				p.stopErr = errors.Join(p.stopErr, ctx.Err())
			}
		}
		if p.supervisor != nil {
			p.stopErr = errors.Join(p.stopErr, p.supervisor.Stop())
		}
		if p.runtime != nil {
			p.stopErr = errors.Join(p.stopErr, p.runtime.Close())
		}
		if p.database != nil {
			p.stopErr = errors.Join(p.stopErr, p.database.Close())
		}
	})
	return p.stopErr
}

var (
	_ http.Handler                         = (*gin.Engine)(nil)
	_ connectedMailboxAuthorizationRuntime = (*runtimegrpc.RuntimePlatformClient)(nil)
)