myxce/src/frontend/apiserver/backend/handler.go

package backend

import (
	"apiserver/models"
	"apiserver/store"
	logger "apiserver/utils"

	"go.uber.org/zap"
)

// DefaultHandler implements NotificationHandler by writing to stores.
type DefaultHandler struct{}

func (h *DefaultHandler) OnJobCreated(job *models.JobInfo) {
	store.Jobs.Put(uint64(job.Unique_JobID), *job)
	DefaultBroker.Publish(Event{Type: EventJobCreated, JobID: uint64(job.Unique_JobID), Status: models.JobStateStr(job.JobStatus), Job: job})
}

func (h *DefaultHandler) OnJobUpdated(job *models.JobInfo) {
	store.Jobs.Put(uint64(job.Unique_JobID), *job)
	DefaultBroker.Publish(Event{Type: EventJobUpdated, JobID: uint64(job.Unique_JobID), Status: models.JobStateStr(job.JobStatus), Job: job})
}

func (h *DefaultHandler) OnJobDeleted(jobid uint64) {
	store.Jobs.Delete(jobid)
	DefaultBroker.Publish(Event{Type: EventJobDeleted, JobID: jobid})
}

func (h *DefaultHandler) OnJobFinished(jobid uint64, exitCode int) {
	existing, ok := store.Jobs.Get(jobid)
	if !ok {
		return
	}
	if exitCode == 0 {
		existing.JobStatus = models.JOBSTAT["DONE"]
	} else {
		existing.JobStatus = models.JOBSTAT["EXIT"]
	}
	store.Jobs.Put(jobid, existing)
	DefaultBroker.Publish(Event{Type: EventJobFinished, JobID: jobid, ExitCode: exitCode})
}

func (h *DefaultHandler) OnQueueUpdated(queue *models.QueueInfo) {
	store.Queues.Put(queue.QueueName, *queue)
}

// OnHostUpdated merges worker-sourced fields into an existing host entry.
// Config-sourced fields (HostModel, HostType, MaxSlot, DisptWindow) are
// preserved from the prior entry if one exists.
func (h *DefaultHandler) OnHostUpdated(incoming *models.HostInfo) {
	merged := *incoming
	if existing, ok := store.Hosts.Get(merged.HostName); ok {
		merged.HostModel = existing.HostModel
		merged.HostType = existing.HostType
		merged.MaxSlot = existing.MaxSlot
		merged.DisptWindow = existing.DisptWindow
	}
	store.Hosts.Put(merged.HostName, merged)
}

// OnHostConfigUpdated applies config-sourced fields onto an existing worker-registered host.
// Hosts not yet registered by a worker are skipped — config entries like "default" (which sets
// cluster-wide defaults) or hosts whose daemons haven't started must not appear in the host list.
func (h *DefaultHandler) OnHostConfigUpdated(incoming *models.HostInfo) {
	existing, ok := store.Hosts.Get(incoming.HostName)
	if !ok {
		return
	}
	existing.HostModel = incoming.HostModel
	existing.HostType = incoming.HostType
	existing.MaxSlot = incoming.MaxSlot
	existing.DisptWindow = incoming.DisptWindow
	store.Hosts.Put(existing.HostName, existing)
}

// OnClusterUpdated applies config-sourced fields onto the existing cluster entry.
// All other fields (built-in/worker-sourced: Version, Status, ControllerName,
// NumServers, Resources, HostTypes, HostModels, ClusterAdmins) are preserved
// from the prior entry.
func (h *DefaultHandler) OnClusterUpdated(incoming *models.ClusterInfo) {
	existing, _ := store.Clusters.Get(incoming.ClusterName)
	existing.ClusterName = incoming.ClusterName
	if incoming.HeartbeatInterval > 0 {
		existing.HeartbeatInterval = incoming.HeartbeatInterval
	}
	existing.Vcpus = incoming.Vcpus
	existing.InactiveHeartbeatCount = incoming.InactiveHeartbeatCount
	store.Clusters.Put(existing.ClusterName, existing)
	SetHeartbeatInterval(incoming.HeartbeatInterval)
}

func (h *DefaultHandler) OnResourceUpdated(res *models.ResourceDefinitionInfo) {
	store.Resources.Put(res.ResourceName, *res)
}

func (h *DefaultHandler) OnResourceLocationUpdated(res *models.ResourceDistributionInfo) {
	key := res.ResourceName + ":" + res.Location
	store.ResourceLocations.Put(key, *res)
}

// dispatchJobArray delivers a locally-expanded compact array push to the
// handler, one call per genuinely-new element (job-array-single-wal-record
// task 5.4, Req 6.2, 6.4, 6.5, 8.4). CGO-free (called from handleDataset's
// XceData_data_jobArray case in clib.go; the elements themselves are already
// decoded by cJobArrayToModels/rawJobArrayToModels before this is called),
// mirroring dispatchConfigPush's split for testability without CGO.
//
// Dedup reuses store.Jobs itself — the SAME jobid-keyed store the ordinary
// single-job push path already writes through (handleDataset's
// XceData_data_jobs case calls OnJobUpdated per job with no separate dedup
// step, relying on the map being keyed by jobid) — checked here BEFORE
// calling the handler so a redelivered push (e.g. after an apiserver
// reconnect) can never clobber a jobid whose state has since moved on.
// This mirrors the scheduler's sched_on_job_array() (handler.c) skip-if-
// exists behavior exactly, not a new dedup mechanism.
//
// Logs exactly one confirmation line (base jobid + array size + new-element
// count) when at least one element is genuinely new; a fully-duplicate
// redelivery (every index already exists) produces zero log lines, matching
// task 5.3's resolved semantics for the equivalent scheduler-side push.
// Returns the count of newly-dispatched elements (test-observable).
func dispatchJobArray(handler NotificationHandler, elements []*models.JobInfo, baseJobID, arraySize uint64) int {
	if handler == nil {
		return 0
	}

	nNew := 0
	for _, job := range elements {
		jobid := uint64(job.Unique_JobID)
		if store.Jobs.Has(jobid) {
			continue
		}
		handler.OnJobUpdated(job)
		nNew++
	}

	if nNew > 0 {
		logger.Info("job array push received and expanded",
			zap.Uint64("base_jobid", baseJobID),
			zap.Uint64("array_size", arraySize),
			zap.Int("new_records", nNew))
	}

	return nNew
}

// dispatchConfigPush delivers converted config-push data to the handler (CGO-free, for testability).
func dispatchConfigPush(
	handler NotificationHandler,
	queues []models.QueueInfo,
	resources []models.ResourceDefinitionInfo,
	hostConfigs []models.HostInfo,
	cluster *models.ClusterInfo,
) {
	for _, q := range queues {
		handler.OnQueueUpdated(&q)
	}
	for _, r := range resources {
		handler.OnResourceUpdated(&r)
	}
	for _, hc := range hostConfigs {
		handler.OnHostConfigUpdated(&hc)
	}
	if cluster != nil {
		handler.OnClusterUpdated(cluster)
	}
}