framework

type BaseJobMaster interface {
	Worker

	// MetaKVClient return business metastore kv client with job-level isolation
	MetaKVClient() metaModel.KVClient

	// MetricFactory return a promethus factory with some underlying labels(e.g. job-id, work-id)
	MetricFactory() promutil.Factory

	// Logger return a zap logger with some underlying fields(e.g. job-id)
	Logger() *zap.Logger

	// GetWorkers return the handle of all workers, from which we can get the worker status、worker id and
	// the method for sending message to specific worker
	GetWorkers() map[frameModel.WorkerID]WorkerHandle

	// CreateWorker requires the framework to dispatch a new worker.
	// If the worker needs to access certain file system resources, it must pass
	// resource ID via CreateWorkerOpt
	CreateWorker(
		workerType frameModel.WorkerType,
		config WorkerConfig,
		opts ...CreateWorkerOpt,
	) (frameModel.WorkerID, error)

	// UpdateJobStatus updates jobmaster(worker of jobmanager) status and
	// sends a 'status updated' message to jobmanager
	UpdateJobStatus(ctx context.Context, status frameModel.WorkerStatus) error

	// CurrentEpoch return the epoch of current job
	CurrentEpoch() frameModel.Epoch

	// SendMessage sends a message of specific topic to jobmanager in a blocking or nonblocking way
	SendMessage(ctx context.Context, topic p2p.Topic, message interface{}, nonblocking bool) error

	// Exit should be called when jobmaster (in user logic) wants to exit.
	// exitReason: ExitReasonFinished/ExitReasonCanceled/ExitReasonFailed
	Exit(ctx context.Context, exitReason ExitReason, err error, detail []byte) error

	// IsMasterReady returns whether the master has received heartbeats for all
	// workers after a fail-over. If this is the first time the JobMaster started up,
	// the return value is always true.
	IsMasterReady() bool

	// IsBaseJobMaster is an empty function used to prevent accidental implementation
	// of this interface.
	IsBaseJobMaster()

	// GetEnabledBucketStorage returns whether the bucket storage is enabled and the corresponding resource type
	// if the bucket exists
	GetEnabledBucketStorage() (bool, resModel.ResourceType)
}

type BaseJobMasterExt interface {
	// TriggerOpenAPIInitialize is used to trigger the initialization of openapi handler.
	// It just delegates to the JobMasterImpl.OnOpenAPIInitialized.
	TriggerOpenAPIInitialize(apiGroup *gin.RouterGroup)

	// IsBaseJobMasterExt is an empty function used to prevent accidental implementation
	// of this interface.
	IsBaseJobMasterExt()
}

type BaseMaster interface {
	Master

	// MetaKVClient return business metastore kv client with job-level isolation
	MetaKVClient() metaModel.KVClient

	// MetricFactory return a promethus factory with some underlying labels(e.g. job-id, work-id)
	MetricFactory() promutil.Factory

	// Logger return a zap logger with some underlying fields(e.g. job-id)
	Logger() *zap.Logger

	// MasterMeta return the meta data of master
	MasterMeta() *frameModel.MasterMeta

	// GetWorkers return the handle of all workers, from which we can get the worker status、worker id and
	// the method for sending message to specific worker
	GetWorkers() map[frameModel.WorkerID]WorkerHandle

	// IsMasterReady returns whether the master has received heartbeats for all
	// workers after a fail-over. If this is the first time the JobMaster started up,
	// the return value is always true.
	IsMasterReady() bool

	// Exit should be called when master (in user logic) wants to exit.
	// exitReason: ExitReasonFinished/ExitReasonCanceled/ExitReasonFailed
	// NOTE: Currently, no implement has used this method, but we still keep it to make the interface intact
	Exit(ctx context.Context, exitReason ExitReason, err error, detail []byte) error

	// CreateWorker is the latest version of CreateWorker, but with
	// a more flexible way of passing options.
	// If the worker needs to access certain file system resources, it must pass
	// resource ID via CreateWorkerOpt
	CreateWorker(
		workerType frameModel.WorkerType,
		config WorkerConfig,
		opts ...CreateWorkerOpt,
	) (frameModel.WorkerID, error)
}

type BaseWorker interface {
	Worker

	// MetaKVClient return business metastore kv client with job-level isolation
	MetaKVClient() metaModel.KVClient

	// MetricFactory return a promethus factory with some underlying labels(e.g. job-id, work-id)
	MetricFactory() promutil.Factory

	// Logger return a zap logger with some underlying fields(e.g. job-id)
	Logger() *zap.Logger

	// UpdateStatus persists the status to framework metastore if worker status is changed and
	// sends 'status updated message' to master.
	UpdateStatus(ctx context.Context, status frameModel.WorkerStatus) error

	// SendMessage sends a message of specific topic to master in a blocking or nonblocking way
	SendMessage(ctx context.Context, topic p2p.Topic, message interface{}, nonblocking bool) error

	// OpenStorage creates a resource and return the resource handle
	OpenStorage(
		ctx context.Context, resourcePath resModel.ResourceID, opts ...broker.OpenStorageOption,
	) (broker.Handle, error)

	// GetEnabledBucketStorage returns whether the bucket storage is enabled and
	// the resource type if bucket exists
	GetEnabledBucketStorage() (bool, resModel.ResourceType)

	// Exit should be called when worker (in user logic) wants to exit.
	// exitReason: ExitReasonFinished/ExitReasonCanceled/ExitReasonFailed
	Exit(ctx context.Context, exitReason ExitReason, err error, extBytes []byte) error
}

type BaseWorkerForTesting struct {
	*DefaultBaseWorker
	Broker *broker.MockBroker
}

type CreateWorkerOpt = master.CreateWorkerOpt

type DefaultBaseJobMaster struct {
	// contains filtered or unexported fields
}

type DefaultBaseMaster struct {
	Impl MasterImpl
	// contains filtered or unexported fields
}

type DefaultBaseWorker struct {
	Impl WorkerImpl
	// contains filtered or unexported fields
}

type ExitReason int

type JobMasterImpl interface {
	MasterImpl

	// OnCancel is triggered when a cancel message is received. It can be
	// triggered multiple times.
	// TODO: when it returns error, framework should close this jobmaster.
	OnCancel(ctx context.Context) error
	// OnOpenAPIInitialized is called as the first callback function of the JobMasterImpl
	// instance, the business logic should only register the OpenAPI handler in it.
	// The implementation must not retain the apiGroup.
	// Note: this function is called before Init().
	// Concurrent safety:
	// - this function is called as the first callback function of an JobMasterImpl
	//   instance, and it's not concurrent with other callbacks.
	OnOpenAPIInitialized(apiGroup *gin.RouterGroup)

	// IsJobMasterImpl is an empty function used to prevent accidental implementation
	// of this interface.
	IsJobMasterImpl()
}

type Master interface {
	Init(ctx context.Context) error
	Poll(ctx context.Context) error
	MasterID() frameModel.MasterID
	Close(ctx context.Context) error
	Stop(ctx context.Context) error
	NotifyExit(ctx context.Context, errIn error) error
}

type MasterFailoverReason struct {
	Code     MasterFailoverReasonCode
	ErrorMsg string
}

type MasterFailoverReasonCode int32

type MasterImpl interface {
	// InitImpl is called at the first time the MasterImpl instance is initialized
	// after OnOpenAPIInitialized. When InitImpl returns without error, framework
	// will try to persist an internal state so further failover will call OnMasterRecovered
	// rather than InitImpl.
	// Return:
	// - error to let the framework call CloseImpl, and framework may retry InitImpl
	//   later for some times. For non-retryable failure, business logic should
	//   call Exit.
	// Concurrent safety:
	// - this function is not concurrent with other callbacks.
	InitImpl(ctx context.Context) error

	// OnMasterRecovered is called when the MasterImpl instance has failover from
	// error by framework. For this MasterImpl instance, it's called after OnOpenAPIInitialized.
	// Return:
	// - error to let the framework call CloseImpl.
	// Concurrent safety:
	// - this function is not concurrent with other callbacks.
	OnMasterRecovered(ctx context.Context) error

	// Tick is called on a fixed interval after MasterImpl's InitImpl or OnMasterRecovered,
	// business logic can do some periodic tasks here.
	// Return:
	// - error to let the framework call CloseImpl.
	// Concurrent safety:
	// - this function may be concurrently called with other callbacks except for
	//   Tick itself, OnOpenAPIInitialized, InitImpl, OnMasterRecovered, CloseImpl,
	//   StopImpl.
	Tick(ctx context.Context) error

	// OnWorkerDispatched is called when the asynchronized action of CreateWorker
	// is finished. Only after OnWorkerDispatched, OnWorkerOnline and OnWorkerStatusUpdated
	// of the same worker may be called.
	// Return:
	// - error to let the framework call CloseImpl.
	// Concurrent safety:
	// - this function may be concurrently called with another worker's OnWorkerXXX,
	//   Tick, CloseImpl, StopImpl, OnCancel.
	OnWorkerDispatched(worker WorkerHandle, result error) error

	// OnWorkerOnline is called when the first heartbeat for a worker is received.
	// NOTE: OnWorkerOffline can appear without OnWorkerOnline
	// Return:
	// - error to let the framework call CloseImpl.
	// Concurrent safety:
	// - this function may be concurrently called with another worker's OnWorkerXXX,
	//   Tick, CloseImpl, StopImpl, OnCancel, the same worker's OnWorkerStatusUpdated.
	OnWorkerOnline(worker WorkerHandle) error

	// OnWorkerOffline is called as the consequence of worker's Exit or heartbeat
	// timed out. It's the last callback function among OnWorkerXXX for a worker.
	// Return:
	// - error to let the framework call CloseImpl.
	// Concurrent safety:
	// - this function may be concurrently called with another worker's OnWorkerXXX,
	//   Tick, CloseImpl, StopImpl, OnCancel.
	OnWorkerOffline(worker WorkerHandle, reason error) error

	// OnWorkerMessage is called when a customized message is received.
	OnWorkerMessage(worker WorkerHandle, topic p2p.Topic, message interface{}) error

	// OnWorkerStatusUpdated is called as the consequence of worker's UpdateStatus.
	// Return:
	// - error to let the framework call CloseImpl.
	// Concurrent safety:
	// - this function may be concurrently called with another worker's OnWorkerXXX,
	//   Tick, CloseImpl, StopImpl, OnCancel, the same worker's OnWorkerOnline.
	OnWorkerStatusUpdated(worker WorkerHandle, newStatus *frameModel.WorkerStatus) error

	// CloseImpl is called as the consequence of returning error from InitImpl,
	// OnMasterRecovered or Tick, the Tick will be stopped after entering this function.
	// And framework may try to create a new masterImpl instance afterwards.
	// Business logic is expected to release resources here, but business developer
	// should be aware that when the runtime is crashed, CloseImpl has no time to
	// be called.
	// TODO: no other callbacks will be called after and concurrent with CloseImpl
	// Concurrent safety:
	// - this function may be concurrently called with OnWorkerMessage, OnCancel,
	//   OnWorkerDispatched, OnWorkerOnline, OnWorkerOffline, OnWorkerStatusUpdated.
	CloseImpl(ctx context.Context)

	// StopImpl is called the consequence of business logic calls Exit. Tick will
	// be stopped after entering this function, and framework will treat this MasterImpl
	// as non-recoverable,
	// There's at most one invocation to StopImpl after Exit. If the runtime is
	// crashed, StopImpl has no time to be called.
	// Concurrent safety:
	// - this function may be concurrently called with OnWorkerMessage, OnCancel,
	//   OnWorkerDispatched, OnWorkerOnline, OnWorkerOffline, OnWorkerStatusUpdated.
	StopImpl(ctx context.Context)
}

type MessageRouter struct {
	// contains filtered or unexported fields
}

type MockHandle = master.MockHandle

type MockMasterImpl struct {
	mock.Mock

	*DefaultBaseMaster
	// contains filtered or unexported fields
}

type MockWorkerHandler struct {
	mock.Mock

	WorkerID frameModel.WorkerID
}

type Worker interface {
	Init(ctx context.Context) error
	Poll(ctx context.Context) error
	ID() runtime.RunnableID
	Close(ctx context.Context) error
	Stop(ctx context.Context) error
	NotifyExit(ctx context.Context, errIn error) error
}

type WorkerConfig = interface{}

type WorkerHandle = master.WorkerHandle

type WorkerImpl interface {
	// InitImpl is called as the consequence of CreateWorker from jobmaster or failover,
	// business logic is expected to do initialization here.
	// Return:
	// - error to let the framework call CloseImpl.
	// Concurrent safety:
	// - this function is called as the first callback function of an WorkerImpl
	//   instance, and it's not concurrent with other callbacks.
	InitImpl(ctx context.Context) error

	// Tick is called on a fixed interval after WorkerImpl is initialized, business
	// logic can do some periodic tasks here.
	// Return:
	// - error to let the framework call CloseImpl.
	// Concurrent safety:
	// - this function may be concurrently called with OnMasterMessage.
	Tick(ctx context.Context) error

	// OnMasterMessage is called when worker receives master message, business developer
	// does not need to implement it.
	// TODO: move it out of WorkerImpl and should not be concurrent with CloseImpl.
	OnMasterMessage(ctx context.Context, topic p2p.Topic, message p2p.MessageValue) error

	// CloseImpl is called as the consequence of returning error from InitImpl or
	// Tick, the Tick will be stopped after entering this function. Business logic
	// is expected to release resources here, but business developer should be aware
	// that when the runtime is crashed, CloseImpl has no time to be called.
	// CloseImpl will only be called for once.
	// TODO: no other callbacks will be called after CloseImpl
	// Concurrent safety:
	// - this function may be concurrently called with OnMasterMessage.
	CloseImpl(ctx context.Context)
}

type WorkerType = frameModel.WorkerType