heicode/heicode/model/agent_callback.go

package model

import (
	"errors"
	"strings"
)

type AgentCallbackEvent struct {
	Id              int    `gorm:"primaryKey" json:"id"`
	EventID         string `gorm:"type:varchar(128);uniqueIndex" json:"event_id"`
	IdempotencyKey  string `gorm:"type:varchar(128);index" json:"idempotency_key"`
	CallbackType    string `gorm:"type:varchar(64);index" json:"callback_type"`
	EventType       string `gorm:"type:varchar(64);index" json:"event_type"`
	DeploymentID    string `gorm:"type:varchar(64);index" json:"deployment_id"`
	SwarmID         string `gorm:"type:varchar(128);index" json:"swarm_id"`
	AgentInstanceID string `gorm:"type:varchar(128);index" json:"agent_instance_id"`
	TaskID          string `gorm:"type:varchar(128);index" json:"task_id"`
	UserID          string `gorm:"type:varchar(64);index" json:"user_id"`
	BindingScope    string `gorm:"type:varchar(512);index" json:"binding_scope"`
	CorrelationID   string `gorm:"type:varchar(64);index" json:"correlation_id"`
	Source          string `gorm:"type:varchar(64)" json:"source"`
	Result          string `gorm:"type:varchar(32)" json:"result"`
	PayloadJSON     string `gorm:"type:text" json:"payload_json"`
	OccurredAt      string `gorm:"type:varchar(32)" json:"occurred_at"`
	CreatedAtMs     int64  `gorm:"bigint;index" json:"created_at_ms"`
}

func (AgentCallbackEvent) TableName() string {
	return "agent_callback_events"
}

type ListAgentCallbackEventsFilter struct {
	DeploymentID  string
	TaskID        string
	CorrelationID string
	Limit         int
}

func InsertAgentCallbackEvent(row *AgentCallbackEvent) (bool, error) {
	if DB == nil || row == nil {
		return false, nil
	}
	if row.EventID == "" {
		return false, errors.New("event_id is required")
	}
	var existing AgentCallbackEvent
	if err := DB.Where("event_id = ?", row.EventID).First(&existing).Error; err == nil {
		return false, nil
	}
	if row.IdempotencyKey != "" {
		if err := DB.Where("idempotency_key = ?", row.IdempotencyKey).First(&existing).Error; err == nil {
			return false, nil
		}
	}
	if err := DB.Create(row).Error; err != nil {
		return false, err
	}
	return true, nil
}

// ListSwarmCallbackEventsAfter returns persisted swarm runtime events for one run
// (matched by deployment_id OR swarm_id) with an id-based `after` cursor for
// incremental polling (#45 events?after). Ordered oldest-first so the client can
// append; the caller uses the last returned Id as the next `after`. Reading from
// HM-persisted callback rows means this needs no live Swarm call.
func ListSwarmCallbackEventsAfter(userID, deploymentID, swarmID string, afterID, limit int) ([]AgentCallbackEvent, error) {
	if DB == nil {
		return nil, nil
	}
	q := DB.Model(&AgentCallbackEvent{})
	switch {
	case deploymentID != "" && swarmID != "":
		q = q.Where("deployment_id = ? OR swarm_id = ?", deploymentID, swarmID)
	case deploymentID != "":
		q = q.Where("deployment_id = ?", deploymentID)
	case swarmID != "":
		q = q.Where("swarm_id = ?", swarmID)
	default:
		return nil, nil
	}
	// 防跨用户泄漏:即使 runtime_swarm_id/deployment_id 碰撞或误写,也按 user_id 收口(#45 复审 #3)。
	if strings.TrimSpace(userID) != "" {
		q = q.Where("user_id = ?", userID)
	}
	if afterID > 0 {
		q = q.Where("id > ?", afterID)
	}
	if limit <= 0 || limit > 1000 {
		limit = 200
	}
	var items []AgentCallbackEvent
	err := q.Order("id asc").Limit(limit).Find(&items).Error
	return items, err
}

func ListAgentCallbackEvents(f ListAgentCallbackEventsFilter) ([]AgentCallbackEvent, error) {
	if DB == nil {
		return nil, nil
	}
	q := DB.Model(&AgentCallbackEvent{})
	if f.DeploymentID != "" {
		q = q.Where("deployment_id = ?", f.DeploymentID)
	}
	if f.TaskID != "" {
		q = q.Where("task_id = ?", f.TaskID)
	}
	if f.CorrelationID != "" {
		q = q.Where("correlation_id = ?", f.CorrelationID)
	}
	limit := f.Limit
	if limit <= 0 || limit > 1000 {
		limit = 200
	}
	var items []AgentCallbackEvent
	err := q.Order("created_at_ms asc, id asc").Limit(limit).Find(&items).Error
	return items, err
}