sundynix-agentix/sundynix-shared/contract/ingest.go

package contract

import "encoding/json"

// 入库工作队列约定（准生产级）。把"解析→切块→向量化→图谱"从网关的裸 goroutine 升级为
// JetStream 持久工作队列：崩溃可重投、有界并发背压、削峰不 OOM。与任务流(SUNDYNIX_TASKS)同级。
const (
	StreamIngest     = "SUNDYNIX_INGEST"     // 入库作业 JetStream 流（持久，作业不因网关重启而丢）
	SubjectIngest    = "sundynix.ingest"     // 入库作业发布前缀；实际 sundynix.ingest.<job_id>
	SubjectIngestAll = "sundynix.ingest.>"   // 流捕获的通配
	ConsumerIngest   = "ingest-workers"      // 入库 worker 持久消费者（队列组：多副本负载均衡 + 背压）
)

// IngestSubject 返回某入库作业的发布主题。
func IngestSubject(jobID string) string { return SubjectIngest + "." + jobID }

// IngestJob 是一条入库作业（claim-check 模式）：大文件原始字节/正文先暂存到对象存储，
// 作业消息只带「暂存键 + 元数据」这条小消息进队列，worker 消费时再按 StageKey 取回原料。
// 这样作业消息恒小（不受 JetStream max_payload 限制），且崩溃重投只搬一个引用。
type IngestJob struct {
	JobID    string `json:"job_id"`
	Owner    string `json:"owner"`              // 雪花 user.id（上传者/创建人，归属）
	SpaceID  string `json:"space_id"`           // 所属工作区（增量3：资源共享作用域）
	KBName   string `json:"kb_name"`            // 知识库展示名（原文留存分区）
	Scoped   string `json:"scoped"`             // space_id/kb 作向量/全文/图谱分区键
	ForceDoc string `json:"force_doc,omitempty"` // 非空=强制文档名（笔记编辑保持身份稳定）
	Filename string `json:"filename,omitempty"` // 非空=文件入库（StageKey 指向原始字节，需先解析）
	StageKey string `json:"stage_key"`          // 暂存对象键：原始文件字节 或 纯文本
	IsText   bool   `json:"is_text"`            // true=StageKey 指向纯文本（跳过解析）
}

// Marshal/Unmarshal 入库作业（JetStream 消息体）。
func (j *IngestJob) Marshal() ([]byte, error) { return json.Marshal(j) }

func UnmarshalIngestJob(data []byte) (*IngestJob, error) {
	var j IngestJob
	if err := json.Unmarshal(data, &j); err != nil {
		return nil, err
	}
	return &j, nil
}