dobotai-platform/internal/store/postgres/audit.go

package postgres

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"

	"gitea.zheromiku.top/Weknora/dobotai-platform/internal/audit"
	"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
)

const listAuditTurnsSQL = `SELECT
	t.id,t.created_at,t.channel,t.status,
	COALESCE(c.conversation_type,''),COALESCE(c.title,''),
	COALESCE(a.external_actor_id,''),COALESCE(rm.actor_name_snapshot,''),
	LEFT(t.question,200),LEFT(t.answer,200),
	CASE WHEN jsonb_typeof(t.references_json)='array' THEN jsonb_array_length(t.references_json) ELSE 0 END,
	COALESCE(am.status,''),COALESCE(am.delivery_method,''),
	COALESCE(fc.up_count,0),COALESCE(fc.down_count,0),COALESCE(fc.unresolved_count,0)
FROM turns t
LEFT JOIN messages rm ON rm.id=t.request_message_id AND rm.tenant_key=t.tenant_key AND rm.channel=t.channel
LEFT JOIN messages am ON am.id=t.answer_message_id AND am.tenant_key=t.tenant_key AND am.channel=t.channel
LEFT JOIN channel_conversations c ON c.id=rm.conversation_id AND c.tenant_key=t.tenant_key AND c.channel=t.channel
LEFT JOIN channel_actors a ON a.id=rm.actor_id AND a.tenant_key=t.tenant_key AND a.channel=t.channel
LEFT JOIN LATERAL (
	SELECT
		(COUNT(*) FILTER (WHERE af.value='up'))::int AS up_count,
		(COUNT(*) FILTER (WHERE af.value='down'))::int AS down_count,
		(COUNT(*) FILTER (WHERE af.value='down' AND af.reason_code=''))::int AS unresolved_count
	FROM answer_feedback af
	WHERE af.tenant_key=t.tenant_key AND af.channel=t.channel AND af.turn_id=t.id
) fc ON TRUE
WHERE t.tenant_key=$1
  AND ($2='' OR t.channel=$2)
  AND ($3='' OR t.status=$3)
  AND ($4::timestamptz IS NULL OR (t.created_at,t.id)<($4,$5))
ORDER BY t.created_at DESC,t.id DESC
LIMIT $6`

const getAuditTurnSQL = `SELECT
	t.id,t.created_at,t.channel,t.status,
	COALESCE(c.conversation_type,''),COALESCE(c.title,''),
	COALESCE(a.external_actor_id,''),COALESCE(rm.actor_name_snapshot,''),
	t.question,t.answer,t.references_json,t.media_json,
	COALESCE(rm.id,''),COALESCE(rm.direction,''),COALESCE(rm.role,''),COALESCE(rm.message_type,''),COALESCE(rm.status,''),COALESCE(rm.actor_name_snapshot,''),rm.created_at,
	COALESCE(am.id,''),COALESCE(am.direction,''),COALESCE(am.role,''),COALESCE(am.message_type,''),COALESCE(am.status,''),am.created_at,am.delivered_at,COALESCE(am.delivery_method,''),COALESCE(am.failure_code,''),
	COALESCE(ff.feedback_json,'[]'::jsonb)
FROM turns t
LEFT JOIN messages rm ON rm.id=t.request_message_id AND rm.tenant_key=t.tenant_key AND rm.channel=t.channel
LEFT JOIN messages am ON am.id=t.answer_message_id AND am.tenant_key=t.tenant_key AND am.channel=t.channel
LEFT JOIN channel_conversations c ON c.id=rm.conversation_id AND c.tenant_key=t.tenant_key AND c.channel=t.channel
LEFT JOIN channel_actors a ON a.id=rm.actor_id AND a.tenant_key=t.tenant_key AND a.channel=t.channel
LEFT JOIN LATERAL (
	SELECT jsonb_agg(jsonb_build_object(
		'actor_external_id',fa.external_actor_id,
		'actor_display_name',fa.display_name,
		'value',af.value,
		'reason_code',af.reason_code,
		'updated_at',af.updated_at
	) ORDER BY af.updated_at DESC,af.id DESC) AS feedback_json
	FROM answer_feedback af
	JOIN channel_actors fa ON fa.id=af.actor_id AND fa.tenant_key=af.tenant_key AND fa.channel=af.channel
	WHERE af.tenant_key=t.tenant_key AND af.channel=t.channel AND af.turn_id=t.id
) ff ON TRUE
WHERE t.tenant_key=$1 AND t.id=$2`

func (s *Store) ListTurns(ctx context.Context, tenantKey string, query audit.ListQuery) ([]audit.TurnSummary, error) {
	var before any
	if query.BeforeCreatedAt != nil {
		before = *query.BeforeCreatedAt
	}
	rows, err := s.pool.Query(ctx, listAuditTurnsSQL, tenantKey, query.Channel, string(query.Status), before, query.BeforeID, query.Limit)
	if err != nil {
		return nil, fmt.Errorf("list audit turns: %w", err)
	}
	defer rows.Close()
	items := make([]audit.TurnSummary, 0)
	for rows.Next() {
		var item audit.TurnSummary
		var status string
		if err := rows.Scan(
			&item.ID, &item.CreatedAt, &item.Channel, &status,
			&item.ConversationType, &item.ConversationTitle,
			&item.ActorExternalID, &item.ActorDisplayName,
			&item.QuestionSummary, &item.AnswerSummary, &item.ReferenceCount,
			&item.AnswerDeliveryStatus, &item.AnswerDeliveryMethod,
			&item.FeedbackUpCount, &item.FeedbackDownCount, &item.FeedbackUnresolvedCount,
		); err != nil {
			return nil, fmt.Errorf("scan audit turn summary: %w", err)
		}
		item.Status = audit.TurnStatus(status)
		items = append(items, item)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate audit turns: %w", err)
	}
	return items, nil
}

func (s *Store) GetTurn(ctx context.Context, tenantKey, turnID string) (audit.TurnDetail, error) {
	var detail audit.TurnDetail
	var status string
	var referencesJSON, mediaJSON, feedbackJSON []byte
	var request audit.MessageMeta
	var answer audit.MessageMeta
	var requestCreated, answerCreated, deliveredAt *time.Time
	err := s.pool.QueryRow(ctx, getAuditTurnSQL, tenantKey, turnID).Scan(
		&detail.ID, &detail.CreatedAt, &detail.Channel, &status,
		&detail.ConversationType, &detail.ConversationTitle,
		&detail.ActorExternalID, &detail.ActorDisplayName,
		&detail.Question, &detail.Answer, &referencesJSON, &mediaJSON,
		&request.ID, &request.Direction, &request.Role, &request.MessageType, &request.Status, &request.ActorName, &requestCreated,
		&answer.ID, &answer.Direction, &answer.Role, &answer.MessageType, &answer.Status, &answerCreated, &deliveredAt, &answer.DeliveryMethod, &answer.FailureCode,
		&feedbackJSON,
	)
	if errors.Is(err, pgx.ErrNoRows) {
		return audit.TurnDetail{}, audit.ErrNotFound
	}
	if err != nil {
		return audit.TurnDetail{}, fmt.Errorf("get audit turn: %w", err)
	}
	detail.Status = audit.TurnStatus(status)

	var references []core.Reference
	if len(referencesJSON) > 0 {
		if err := json.Unmarshal(referencesJSON, &references); err != nil {
			return audit.TurnDetail{}, fmt.Errorf("decode audit references: %w", err)
		}
	}
	detail.References = safeAuditReferences(references)
	detail.ReferenceCount = len(detail.References)

	var media []core.MediaAudit
	if len(mediaJSON) > 0 {
		if err := json.Unmarshal(mediaJSON, &media); err != nil {
			return audit.TurnDetail{}, fmt.Errorf("decode audit media: %w", err)
		}
	}
	detail.Media = safeAuditMedia(media)
	if err := json.Unmarshal(feedbackJSON, &detail.Feedback); err != nil {
		return audit.TurnDetail{}, fmt.Errorf("decode audit feedback: %w", err)
	}

	if request.ID != "" {
		if requestCreated != nil {
			request.CreatedAt = *requestCreated
		}
		detail.RequestMessage = &request
	}
	if answer.ID != "" {
		if answerCreated != nil {
			answer.CreatedAt = *answerCreated
		}
		answer.DeliveredAt = deliveredAt
		detail.AnswerMessage = &answer
		detail.AnswerDeliveryStatus = answer.Status
		detail.AnswerDeliveryMethod = answer.DeliveryMethod
	}
	return detail, nil
}

func safeAuditReferences(references []core.Reference) []audit.SafeReference {
	if len(references) == 0 {
		return nil
	}
	safe := make([]audit.SafeReference, 0, len(references))
	for _, reference := range references {
		safe = append(safe, audit.SafeReference{
			ID: reference.ID, KnowledgeID: reference.KnowledgeID,
			KnowledgeTitle: reference.KnowledgeTitle, Score: reference.Score,
		})
	}
	return safe
}

func safeAuditMedia(media []core.MediaAudit) []audit.SafeMedia {
	if len(media) == 0 {
		return nil
	}
	safe := make([]audit.SafeMedia, 0, len(media))
	for _, item := range media {
		safe = append(safe, audit.SafeMedia{
			Type: item.Type, MIMEType: item.MIMEType, SizeBytes: item.SizeBytes,
			Width: item.Width, Height: item.Height,
		})
	}
	return safe
}

var _ audit.TurnRepository = (*Store)(nil)