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

package postgres

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"time"

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

var (
	errRowsAffected   = errors.New("support notification: RowsAffected != 1")
	errBadFailureCode = errors.New("support notification: only dingtalk_group_send_failed is accepted")
	errEmptyCaseID    = errors.New("support notification: caseID is required")
)

var _ coverage.NotificationRepository = (*Store)(nil)

// Claim implements coverage.NotificationRepository. It atomically claims up to
// limit pending/retryable/expired-notifying support cases using a single CTE
// with FOR UPDATE SKIP LOCKED.
func (s *Store) Claim(ctx context.Context, limit int, maxAttempts int, staleBefore time.Time) ([]coverage.PendingSupportCase, error) {
	if s == nil || s.pool == nil {
		return nil, coverage.ErrRepository
	}

	rows, err := s.pool.Query(ctx, `
		WITH candidates AS (
			SELECT sc.id
			FROM support_cases sc
			WHERE (sc.status = 'pending' OR sc.status = 'notification_failed'
			       OR (sc.status = 'notifying' AND sc.last_notification_at < $3))
			  AND sc.faq_export_status = 'exported'
			  AND sc.notification_attempts < $2
			ORDER BY sc.created_at, sc.id
			LIMIT $1
			FOR UPDATE SKIP LOCKED
		),
		claimed AS (
			UPDATE support_cases sc
			SET status = 'notifying',
			    notification_attempts = sc.notification_attempts + 1,
			    last_notification_at = NOW(),
			    last_notification_failure_code = '',
			    updated_at = NOW()
			FROM candidates c
			WHERE sc.id = c.id
			RETURNING sc.id, sc.tenant_key, sc.channel, sc.turn_id,
			          sc.notification_attempts, sc.created_at, sc.faq_record_id, sc.trigger_type
		)
		SELECT
			cl.id,
			cl.tenant_key,
			cl.channel,
			cl.turn_id,
			cl.notification_attempts,
			COALESCE(t.question, '') AS question,
			COALESCE(m.actor_name_snapshot, '') AS sender_name,
			COALESCE(c.conversation_type, 'direct') AS conversation_type,
			m.created_at AS asked_at,
			COALESCE(history.previous_questions, ARRAY[]::TEXT[]) AS previous_questions,
			cl.faq_record_id,
			cl.trigger_type
		FROM claimed cl
		JOIN turns t ON t.id = cl.turn_id
			AND t.tenant_key = cl.tenant_key
			AND t.channel = cl.channel
		JOIN messages m ON m.id = t.request_message_id
			AND m.tenant_key = cl.tenant_key
			AND m.channel = cl.channel
		LEFT JOIN channel_conversations c ON c.id = m.conversation_id
			AND c.tenant_key = cl.tenant_key
			AND c.channel = cl.channel
		LEFT JOIN LATERAL (
			SELECT ARRAY_AGG(previous.question ORDER BY previous.created_at DESC) AS previous_questions
			FROM (
				SELECT t2.question, t2.created_at
				FROM turns t2
				JOIN messages m2 ON m2.id = t2.request_message_id
					AND m2.tenant_key = t2.tenant_key
					AND m2.channel = t2.channel
				WHERE t2.tenant_key = cl.tenant_key
					AND t2.channel = cl.channel
					AND m2.actor_id = m.actor_id
					AND t2.created_at < t.created_at
				ORDER BY t2.created_at DESC
				LIMIT 5
			) previous
		) history ON TRUE
		ORDER BY cl.created_at, cl.id`,
		limit, maxAttempts, staleBefore)
	if err != nil {
		return nil, fmt.Errorf("claim support cases: %w", err)
	}
	defer rows.Close()

	var cases []coverage.PendingSupportCase
	for rows.Next() {
		var (
			id                string
			tenantKey         string
			channel           string
			turnID            string
			attempts          int
			question          string
			senderName        string
			conversationType  string
			askedAt           time.Time
			previousQuestions []string
			faqRecordID       string
			triggerType       string
		)
		if err := rows.Scan(&id, &tenantKey, &channel, &turnID, &attempts,
			&question, &senderName, &conversationType, &askedAt,
			&previousQuestions, &faqRecordID, &triggerType); err != nil {
			rows.Close()
			return nil, fmt.Errorf("claim scan: %w", err)
		}
		cases = append(cases, coverage.PendingSupportCase{
			SupportNotification: coverage.SupportNotification{
				CaseID:            id,
				TriggerType:       triggerType,
				Question:          question,
				SenderName:        senderName,
				Channel:           channel,
				ConversationType:  core.ConversationType(conversationType),
				AskedAt:           askedAt,
				PreviousQuestions: previousQuestions,
				FAQRecordID:       faqRecordID,
			},
			Attempts: attempts,
		})
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("claim rows: %w", err)
	}
	return cases, nil
}

// MarkNotified implements coverage.NotificationRepository.
// It transitions a case from notifying to notified. RowsAffected must be exactly 1.
func (s *Store) MarkNotified(ctx context.Context, caseID string) error {
	if strings.TrimSpace(caseID) == "" {
		return errEmptyCaseID
	}
	if s == nil || s.pool == nil {
		return coverage.ErrRepository
	}

	tag, err := s.pool.Exec(ctx, `
		UPDATE support_cases
		SET status = 'notified',
		    notified_at = NOW(),
		    last_notification_failure_code = '',
		    updated_at = NOW()
		WHERE id = $1 AND status = 'notifying'`,
		caseID)
	if err != nil {
		return fmt.Errorf("mark notified: %w", err)
	}
	if tag.RowsAffected() != 1 {
		return errRowsAffected
	}
	return nil
}

// MarkFailed implements coverage.NotificationRepository.
// It transitions a case from notifying to notification_failed with a stable
// failure code. Only "dingtalk_group_send_failed" is accepted. Raw errors
// must never be persisted. RowsAffected must be exactly 1.
func (s *Store) MarkFailed(ctx context.Context, caseID string, failureCode string) error {
	if strings.TrimSpace(caseID) == "" {
		return errEmptyCaseID
	}
	if failureCode != coverage.FailureCodeDingtalkGroupSendFailed {
		return errBadFailureCode
	}
	if s == nil || s.pool == nil {
		return coverage.ErrRepository
	}

	tag, err := s.pool.Exec(ctx, `
		UPDATE support_cases
		SET status = 'notification_failed',
		    last_notification_failure_code = $2,
		    updated_at = NOW()
		WHERE id = $1 AND status = 'notifying'`,
		caseID, failureCode)
	if err != nil {
		return fmt.Errorf("mark failed: %w", err)
	}
	if tag.RowsAffected() != 1 {
		return errRowsAffected
	}
	return nil
}