Function sendAgentCommand

server-source-code/internal/queue/jobs.go:405–450

sendAgentCommand is a helper that sends a JSON command to the agent and updates job_history.

(ctx context.Context, h *ReportNowHandler, p ReportNowPayload, msgType, taskID string, retryCount int)

// sendAgentCommand is a helper that sends a JSON command to the agent and updates job_history.
func sendAgentCommand(ctx context.Context, h *ReportNowHandler, p ReportNowPayload, msgType, taskID string, retryCount int) error {
	taskIDVal, _ := asynq.GetTaskID(ctx)
	if taskID == "" {
		taskID = taskIDVal
	}
	attempt := int32(retryCount + 1)

	if h.db != nil && taskID != "" && retryCount == 0 {
		host, err := h.db.Queries.GetHostByApiID(ctx, p.ApiID)
		var hostID *string
		if err == nil {
			hostID = &host.ID
		}
		apiIDPtr := &p.ApiID
		_ = h.db.Queries.InsertJobHistory(ctx, db.InsertJobHistoryParams{
			ID:            uuid.New().String(),
			JobID:         taskID,
			QueueName:     QueueAgentCommands,
			JobName:       msgType,
			HostID:        hostID,
			ApiID:         apiIDPtr,
			Status:        "active",
			AttemptNumber: attempt,
		})
	}

	if !h.registry.IsConnected(p.ApiID) {
		h.log.Warn(msgType+": agent not connected", "api_id", p.ApiID)
		if taskID != "" && h.db != nil {
			msg := "Agent not connected"
			_ = h.db.Queries.UpdateJobHistoryFailed(ctx, db.UpdateJobHistoryFailedParams{JobID: taskID, ErrorMessage: &msg})
		}
		return nil
	}
	msg := []byte(`{"type":"` + msgType + `"}`)
	if err := h.registry.SendMessage(p.ApiID, websocket.TextMessage, msg); err != nil {
		h.log.Warn(msgType+": write failed", "api_id", p.ApiID, "error", err)
		return err
	}

	if taskID != "" && h.db != nil {
		_ = h.db.Queries.UpdateJobHistoryCompleted(ctx, taskID)
	}
	h.log.Info(msgType+" sent", "api_id", p.ApiID)
	return nil
}

// RefreshIntegrationStatusHandler handles refresh_integration_status jobs.
type RefreshIntegrationStatusHandler struct {