Agent-System/internal/artifactvault/ingest.go

package artifactvault

import (
	"context"
	"fmt"
	"strconv"
	"time"

	"chat-agent-system/internal/app/uow"

	"github.com/jackc/pgx/v5"
)

// Ingest is the single write path from domain terminals into staging/outbox.
type Ingest struct {
	Env string
	Reg Registry
}

func ProjectKeyFromID(projectID int64) string {
	if projectID <= 0 {
		return "unscoped"
	}
	return strconv.FormatInt(projectID, 10)
}

func (i *Ingest) Request(ctx context.Context, facts Facts, req PublishRequest) error {
	if i == nil || i.Reg == nil {
		return nil
	}
	if req.Environment == "" {
		req.Environment = i.Env
	}
	if req.ProjectExternalKey == "" {
		req.ProjectExternalKey = ProjectKeyFromID(req.ProjectID)
	}
	if req.PolicyVersion == "" {
		req.PolicyVersion = PolicyVersion
	}
	if req.Classification == "" {
		req.Classification = ClassifyPayload(req.Payload)
	}
	facts.Payload = req.Payload
	facts.Classification = req.Classification
	_, _, err := i.Reg.Request(ctx, facts, req)
	return err
}

func ApprovedFacts(docState string, approvedAt *time.Time, snapshotID int64) Facts {
	return Facts{
		ArtifactType:     TypeApprovedDocument,
		DocState:         docState,
		ApprovedAt:       approvedAt,
		ReviewSnapshotID: snapshotID,
	}
}

func AgentCommitFacts(outputSnapshotID int64) Facts {
	return Facts{
		ArtifactType:       TypeAgentCommit,
		CandidateStatus:    "committed",
		OutputSnapshotID:   outputSnapshotID,
		ChangeKind:         "agent_commit",
		ChangeToSnapshotID: outputSnapshotID,
	}
}

func SealedFacts(sealedAt *time.Time, headSnapshotID int64) Facts {
	return Facts{
		ArtifactType:   TypeSealedVersion,
		VersionStatus:  "sealed",
		SealedAt:       sealedAt,
		HeadSnapshotID: headSnapshotID,
	}
}

func AssistantFacts(turnStatus string, completedAt *time.Time, msgID int64, policy string) Facts {
	return Facts{
		ArtifactType:       TypeAssistantReply,
		ChatTurnStatus:     turnStatus,
		CompletedAt:        completedAt,
		AssistantMessageID: msgID,
		ExportPolicy:       policy,
	}
}

func FreeformFacts(explicit bool, hasSnapshot bool) Facts {
	return Facts{
		ArtifactType:         TypeFreeformPublish,
		ExplicitPublish:      explicit,
		HasImmutableSnapshot: hasSnapshot,
	}
}

func (i *Ingest) AgentCommit(ctx context.Context, projectID int64, docID, eventID string, snapshotID int64, payload []byte) error {
	return i.Request(ctx, AgentCommitFacts(snapshotID), PublishRequest{
		ArtifactType: TypeAgentCommit, Environment: i.Env, ProjectID: projectID,
		LogicalResourceID: docID, SourceEventKind: "agent_commit", SourceEventID: eventID,
		SourceSnapshotID: snapshotID, Payload: payload, TerminalState: "committed",
	})
}

func (i *Ingest) SealedVersion(ctx context.Context, projectID int64, docID, eventID string, snapshotID int64, sealedAt *time.Time, payload []byte) error {
	return i.Request(ctx, SealedFacts(sealedAt, snapshotID), PublishRequest{
		ArtifactType: TypeSealedVersion, Environment: i.Env, ProjectID: projectID,
		LogicalResourceID: docID + ":sealed", SourceEventKind: "version_sealed", SourceEventID: eventID,
		SourceSnapshotID: snapshotID, Payload: payload, TerminalState: "sealed",
	})
}

func (i *Ingest) Approved(ctx context.Context, projectID int64, docID string, snapshotID int64, approvedAt *time.Time, payload []byte) error {
	return i.Request(ctx, ApprovedFacts("approved", approvedAt, snapshotID), PublishRequest{
		ArtifactType: TypeApprovedDocument, Environment: i.Env, ProjectID: projectID,
		LogicalResourceID: docID, SourceEventKind: "approved", SourceEventID: fmt.Sprintf("approved:%s:%d", docID, snapshotID),
		SourceSnapshotID: snapshotID, Payload: payload, TerminalState: "approved",
	})
}

func (i *Ingest) FreeformPublish(ctx context.Context, projectID int64, docID, eventID string, snapshotID int64, payload []byte) error {
	return i.Request(ctx, FreeformFacts(true, snapshotID > 0 || len(payload) > 0), PublishRequest{
		ArtifactType: TypeFreeformPublish, Environment: i.Env, ProjectID: projectID,
		LogicalResourceID: docID, SourceEventKind: "freeform_publish", SourceEventID: eventID,
		SourceSnapshotID: snapshotID, Payload: payload, TerminalState: "published",
	})
}

func (i *Ingest) requestTX(ctx context.Context, tx pgx.Tx, facts Facts, req PublishRequest) error {
	if i == nil {
		return nil
	}
	if req.Environment == "" {
		req.Environment = i.Env
	}
	if req.ProjectExternalKey == "" {
		req.ProjectExternalKey = ProjectKeyFromID(req.ProjectID)
	}
	if req.PolicyVersion == "" {
		req.PolicyVersion = PolicyVersion
	}
	if req.Classification == "" {
		req.Classification = ClassifyPayload(req.Payload)
	}
	facts.Payload = req.Payload
	facts.Classification = req.Classification
	_, _, err := RequestWithTx(ctx, tx, facts, req)
	return err
}

func (i *Ingest) StageAgentCommit(ctx context.Context, projectID int64, docID, eventID string, snapshotID int64, payload []byte) error {
	scope, ok := uow.ScopeFrom(ctx)
	if !ok {
		return fmt.Errorf("artifactvault: staging requires a unit of work")
	}
	tx, err := uow.AsPG(scope)
	if err != nil {
		return err
	}
	return i.AgentCommitTX(ctx, tx, projectID, docID, eventID, snapshotID, payload)
}

func (i *Ingest) AgentCommitTX(ctx context.Context, tx pgx.Tx, projectID int64, docID, eventID string, snapshotID int64, payload []byte) error {
	return i.requestTX(ctx, tx, AgentCommitFacts(snapshotID), PublishRequest{
		ArtifactType: TypeAgentCommit, Environment: i.Env, ProjectID: projectID,
		LogicalResourceID: docID, SourceEventKind: "agent_commit", SourceEventID: eventID,
		SourceSnapshotID: snapshotID, Payload: payload, TerminalState: "committed",
	})
}

func (i *Ingest) SealedVersionTX(ctx context.Context, tx pgx.Tx, projectID int64, docID, eventID string, snapshotID int64, sealedAt *time.Time, payload []byte) error {
	return i.requestTX(ctx, tx, SealedFacts(sealedAt, snapshotID), PublishRequest{
		ArtifactType: TypeSealedVersion, Environment: i.Env, ProjectID: projectID,
		LogicalResourceID: docID + ":sealed", SourceEventKind: "version_sealed", SourceEventID: eventID,
		SourceSnapshotID: snapshotID, Payload: payload, TerminalState: "sealed",
	})
}

func (i *Ingest) ApprovedTX(ctx context.Context, tx pgx.Tx, projectID int64, docID string, snapshotID int64, approvedAt *time.Time, payload []byte) error {
	return i.requestTX(ctx, tx, ApprovedFacts("approved", approvedAt, snapshotID), PublishRequest{
		ArtifactType: TypeApprovedDocument, Environment: i.Env, ProjectID: projectID,
		LogicalResourceID: docID, SourceEventKind: "approved", SourceEventID: fmt.Sprintf("approved:%s:%d", docID, snapshotID),
		SourceSnapshotID: snapshotID, Payload: payload, TerminalState: "approved",
	})
}

func (i *Ingest) AssistantReply(ctx context.Context, projectID int64, logical, eventID string, msgID int64, runID, policy string, completedAt *time.Time, payload []byte) error {
	return i.Request(ctx, AssistantFacts("completed", completedAt, msgID, policy), PublishRequest{
		ArtifactType: TypeAssistantReply, Environment: i.Env, ProjectID: projectID,
		LogicalResourceID: logical, SourceEventKind: "assistant_completed", SourceEventID: eventID,
		SourceMessageID: msgID, SourceRunID: runID, Payload: payload, TerminalState: "completed",
	})
}

func (i *Ingest) AssistantReplyPin(ctx context.Context, projectID int64, logical, eventID string, msgID int64, runID, policy string, completedAt *time.Time, payload []byte) error {
	facts := AssistantFacts("completed", completedAt, msgID, policy)
	facts.ExplicitPin = true
	facts.PinAfterCompleted = true
	return i.Request(ctx, facts, PublishRequest{
		ArtifactType: TypeAssistantReply, Environment: i.Env, ProjectID: projectID,
		LogicalResourceID: logical, SourceEventKind: "assistant_pin", SourceEventID: eventID,
		SourceMessageID: msgID, SourceRunID: runID, Payload: payload, TerminalState: "completed",
	})
}