geo/server/internal/tenant/app/media_service.go

package app

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"strings"
	"time"

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

	"github.com/geo-platform/tenant-api/internal/shared/auth"
	"github.com/geo-platform/tenant-api/internal/shared/response"
)

type MediaService struct {
	pool *pgxpool.Pool
}

func NewMediaService(pool *pgxpool.Pool) *MediaService {
	return &MediaService{pool: pool}
}

type MediaPlatformResponse struct {
	ID          int64   `json:"id"`
	PlatformID  string  `json:"platform_id"`
	Name        string  `json:"name"`
	Category    string  `json:"category"`
	ShortName   string  `json:"short_name"`
	AccentColor string  `json:"accent_color"`
	LoginURL    *string `json:"login_url"`
	LogoURL     *string `json:"logo_url"`
	Status      string  `json:"status"`
	SortOrder   int     `json:"sort_order"`
}

type PublishRecordResponse struct {
	ID                 int64      `json:"id"`
	PublishBatchID     int64      `json:"publish_batch_id"`
	ArticleID          int64      `json:"article_id"`
	ArticleTitle       *string    `json:"article_title,omitempty"`
	PlatformAccountID  *int64     `json:"platform_account_id"`
	DesktopAccountID   *string    `json:"desktop_account_id"`
	DesktopTaskID      *string    `json:"desktop_task_id,omitempty"`
	TargetType         string     `json:"target_type"`
	TargetConnectionID *int64     `json:"target_connection_id"`
	PlatformID         string     `json:"platform_id"`
	PlatformName       string     `json:"platform_name"`
	PlatformNickname   string     `json:"platform_nickname"`
	Status             string     `json:"status"`
	ExternalArticleID  *string    `json:"external_article_id"`
	ExternalArticleURL *string    `json:"external_article_url"`
	ExternalManageURL  *string    `json:"external_manage_url"`
	PublishedAt        *time.Time `json:"published_at"`
	ErrorMessage       *string    `json:"error_message"`
	CreatedAt          time.Time  `json:"created_at"`
	UpdatedAt          time.Time  `json:"updated_at"`
}

type ListPublishRecordsRequest struct {
	Page     int
	PageSize int
	Title    string
}

type PublishRecordListResponse struct {
	Items        []PublishRecordResponse `json:"items"`
	Page         int                     `json:"page"`
	PageSize     int                     `json:"page_size"`
	Total        int                     `json:"total"`
	PendingCount int                     `json:"pending_count"`
	HistoryTotal int                     `json:"history_total"`
}

func (s *MediaService) ListPlatforms(ctx context.Context) ([]MediaPlatformResponse, error) {
	rows, err := s.pool.Query(ctx, `
		SELECT id, platform_id, name, category, short_name, accent_color, login_url, logo_url, status, sort_order
		FROM media_platforms
		WHERE status = 'active'
		ORDER BY sort_order ASC, id ASC
	`)
	if err != nil {
		return nil, response.ErrInternal(50030, "media_platform_query_failed", "failed to list media platforms")
	}
	defer rows.Close()

	items := make([]MediaPlatformResponse, 0)
	for rows.Next() {
		var item MediaPlatformResponse
		if err := rows.Scan(
			&item.ID,
			&item.PlatformID,
			&item.Name,
			&item.Category,
			&item.ShortName,
			&item.AccentColor,
			&item.LoginURL,
			&item.LogoURL,
			&item.Status,
			&item.SortOrder,
		); err != nil {
			return nil, response.ErrInternal(50030, "media_platform_scan_failed", err.Error())
		}
		items = append(items, item)
	}
	return items, nil
}

func (s *MediaService) ListPublishRecords(ctx context.Context, req ListPublishRecordsRequest) (*PublishRecordListResponse, error) {
	actor := auth.MustActor(ctx)
	brandID, err := requireCurrentBrandID(ctx)
	if err != nil {
		return nil, err
	}

	page := req.Page
	if page <= 0 {
		page = 1
	}
	pageSize := req.PageSize
	if pageSize <= 0 || pageSize > 50 {
		pageSize = 10
	}
	title := strings.TrimSpace(req.Title)
	pendingStatuses := []string{"queued", "publishing"}
	historyStatuses := []string{"success", "failed", "cancelled", "canceled"}

	pendingItems, err := s.listPublishRecordsByStatuses(ctx, actor.TenantID, brandID, pendingStatuses, title, 0, 0, "pr.created_at ASC, pr.id ASC")
	if err != nil {
		return nil, err
	}
	historyTotal, err := s.countPublishRecordsByStatuses(ctx, actor.TenantID, brandID, historyStatuses, title)
	if err != nil {
		return nil, err
	}

	offset := (page - 1) * pageSize
	items := make([]PublishRecordResponse, 0, pageSize)
	total := len(pendingItems) + historyTotal

	if offset < len(pendingItems) {
		end := offset + pageSize
		if end > len(pendingItems) {
			end = len(pendingItems)
		}
		items = append(items, pendingItems[offset:end]...)
	}

	remaining := pageSize - len(items)
	historyOffset := offset - len(pendingItems)
	if historyOffset < 0 {
		historyOffset = 0
	}
	if remaining > 0 && historyOffset < historyTotal {
		historyItems, listErr := s.listPublishRecordsByStatuses(ctx, actor.TenantID, brandID, historyStatuses, title, remaining, historyOffset, "pr.updated_at DESC, pr.id DESC")
		if listErr != nil {
			return nil, listErr
		}
		items = append(items, historyItems...)
	}

	return &PublishRecordListResponse{
		Items:        items,
		Page:         page,
		PageSize:     pageSize,
		Total:        total,
		PendingCount: len(pendingItems),
		HistoryTotal: historyTotal,
	}, nil
}

func (s *MediaService) listPublishRecordsByStatuses(
	ctx context.Context,
	tenantID int64,
	brandID int64,
	statuses []string,
	title string,
	limit int,
	offset int,
	orderBy string,
) ([]PublishRecordResponse, error) {
	query := `
		SELECT pr.id, pr.publish_batch_id, pr.article_id,
		       COALESCE(av.title, NULLIF(a.wizard_state_json ->> 'title', '')) AS article_title,
		       pr.platform_account_id, pa.desktop_id::text, dt.desktop_id::text,
		       COALESCE(pr.target_type, 'platform_account'), pr.target_connection_id, pr.platform_id,
		       COALESCE(mp.name, esc.site_name, esc.name, pr.platform_id) AS platform_name,
		       CASE
		         WHEN COALESCE(pr.target_type, 'platform_account') = 'enterprise_site'
		           THEN COALESCE(esc.site_name, esc.name, '企业站点')
		         ELSE COALESCE(pa.nickname, '')
		       END AS platform_nickname,
		       pr.status, pr.external_article_id, pr.external_article_url,
		       pr.external_manage_url, pr.published_at, pr.error_message, pr.created_at, pr.updated_at
		FROM publish_records pr
		JOIN articles a ON a.id = pr.article_id AND a.tenant_id = pr.tenant_id
		LEFT JOIN article_versions av ON av.id = a.current_version_id AND av.article_id = a.id
		LEFT JOIN media_platforms mp ON mp.platform_id = pr.platform_id
		LEFT JOIN platform_accounts pa ON pa.id = pr.platform_account_id
		LEFT JOIN enterprise_site_connections esc
		  ON esc.id = pr.target_connection_id
		 AND esc.tenant_id = pr.tenant_id
		LEFT JOIN LATERAL (
			SELECT desktop_id
			FROM desktop_tasks
			WHERE tenant_id = pr.tenant_id
			  AND kind = 'publish'
			  AND payload->>'publish_record_id' = pr.id::text
			ORDER BY updated_at DESC, desktop_id DESC
			LIMIT 1
		) dt ON TRUE
		WHERE pr.tenant_id = $1
		  AND a.brand_id = $2
		  AND a.deleted_at IS NULL
		  AND pr.status = ANY($3)
		  AND (
		    $4 = ''
		    OR COALESCE(av.title, NULLIF(a.wizard_state_json ->> 'title', ''), '') ILIKE '%' || $4 || '%'
		  )
	`
	args := []any{tenantID, brandID, statuses, title}

	if strings.TrimSpace(orderBy) != "" {
		query += "\nORDER BY " + orderBy
	}
	if limit > 0 {
		query += "\nLIMIT $5 OFFSET $6"
		args = append(args, limit, offset)
	}

	rows, err := s.pool.Query(ctx, query, args...)
	if err != nil {
		return nil, response.ErrInternal(50043, "publish_record_query_failed", "failed to list publish records")
	}
	defer rows.Close()

	items := make([]PublishRecordResponse, 0)
	for rows.Next() {
		var item PublishRecordResponse
		var desktopAccountID *string
		var desktopTaskID *string
		if err := rows.Scan(
			&item.ID,
			&item.PublishBatchID,
			&item.ArticleID,
			&item.ArticleTitle,
			&item.PlatformAccountID,
			&desktopAccountID,
			&desktopTaskID,
			&item.TargetType,
			&item.TargetConnectionID,
			&item.PlatformID,
			&item.PlatformName,
			&item.PlatformNickname,
			&item.Status,
			&item.ExternalArticleID,
			&item.ExternalArticleURL,
			&item.ExternalManageURL,
			&item.PublishedAt,
			&item.ErrorMessage,
			&item.CreatedAt,
			&item.UpdatedAt,
		); err != nil {
			return nil, response.ErrInternal(50043, "publish_record_scan_failed", err.Error())
		}
		if desktopAccountID != nil && strings.TrimSpace(*desktopAccountID) != "" {
			item.DesktopAccountID = desktopAccountID
		}
		if desktopTaskID != nil && strings.TrimSpace(*desktopTaskID) != "" {
			item.DesktopTaskID = desktopTaskID
		}
		normalizePublishRecordForResponse(&item)
		items = append(items, item)
	}
	if err := rows.Err(); err != nil {
		return nil, response.ErrInternal(50043, "publish_record_query_failed", "failed to list publish records")
	}
	return items, nil
}

func (s *MediaService) countPublishRecordsByStatuses(
	ctx context.Context,
	tenantID int64,
	brandID int64,
	statuses []string,
	title string,
) (int, error) {
	var count int
	err := s.pool.QueryRow(ctx, `
		SELECT COUNT(1)
		FROM publish_records pr
		JOIN articles a ON a.id = pr.article_id AND a.tenant_id = pr.tenant_id
		LEFT JOIN article_versions av ON av.id = a.current_version_id AND av.article_id = a.id
		WHERE pr.tenant_id = $1
		  AND a.brand_id = $2
		  AND a.deleted_at IS NULL
		  AND pr.status = ANY($3)
		  AND (
		    $4 = ''
		    OR COALESCE(av.title, NULLIF(a.wizard_state_json ->> 'title', ''), '') ILIKE '%' || $4 || '%'
		  )
	`, tenantID, brandID, statuses, title).Scan(&count)
	if err != nil {
		return 0, response.ErrInternal(50043, "publish_record_query_failed", "failed to list publish records")
	}
	return count, nil
}

func (s *MediaService) ListArticlePublishRecords(ctx context.Context, articleID int64) ([]PublishRecordResponse, error) {
	actor := auth.MustActor(ctx)
	brandID, err := requireCurrentBrandID(ctx)
	if err != nil {
		return nil, err
	}
	return s.listArticlePublishRecords(ctx, actor.TenantID, brandID, articleID)
}

func (s *MediaService) listArticlePublishRecords(ctx context.Context, tenantID, brandID, articleID int64) ([]PublishRecordResponse, error) {
	rows, err := s.pool.Query(ctx, `
		SELECT pr.id, pr.publish_batch_id, pr.article_id, pr.platform_account_id, pa.desktop_id::text,
		       COALESCE(pr.target_type, 'platform_account'), pr.target_connection_id, pr.platform_id,
		       COALESCE(mp.name, esc.site_name, esc.name, pr.platform_id) AS platform_name,
		       CASE
		         WHEN COALESCE(pr.target_type, 'platform_account') = 'enterprise_site'
		           THEN COALESCE(esc.site_name, esc.name, '企业站点')
		         ELSE COALESCE(pa.nickname, '')
		       END AS platform_nickname,
		       pr.status, pr.external_article_id, pr.external_article_url,
		       pr.external_manage_url, pr.published_at, pr.error_message, pr.created_at, pr.updated_at
		FROM publish_records pr
		LEFT JOIN media_platforms mp ON mp.platform_id = pr.platform_id
		LEFT JOIN platform_accounts pa ON pa.id = pr.platform_account_id
		LEFT JOIN enterprise_site_connections esc
		  ON esc.id = pr.target_connection_id
		 AND esc.tenant_id = pr.tenant_id
		JOIN articles a ON a.id = pr.article_id AND a.tenant_id = pr.tenant_id
		WHERE pr.tenant_id = $1 AND pr.article_id = $2 AND a.brand_id = $3 AND a.deleted_at IS NULL
		ORDER BY pr.created_at DESC, pr.id DESC
	`, tenantID, articleID, brandID)
	if err != nil {
		return nil, response.ErrInternal(50043, "publish_record_query_failed", "failed to list publish records")
	}
	defer rows.Close()

	items := make([]PublishRecordResponse, 0)
	for rows.Next() {
		var item PublishRecordResponse
		var desktopAccountID *string
		if err := rows.Scan(
			&item.ID,
			&item.PublishBatchID,
			&item.ArticleID,
			&item.PlatformAccountID,
			&desktopAccountID,
			&item.TargetType,
			&item.TargetConnectionID,
			&item.PlatformID,
			&item.PlatformName,
			&item.PlatformNickname,
			&item.Status,
			&item.ExternalArticleID,
			&item.ExternalArticleURL,
			&item.ExternalManageURL,
			&item.PublishedAt,
			&item.ErrorMessage,
			&item.CreatedAt,
			&item.UpdatedAt,
		); err != nil {
			return nil, response.ErrInternal(50043, "publish_record_scan_failed", err.Error())
		}
		if desktopAccountID != nil && strings.TrimSpace(*desktopAccountID) != "" {
			item.DesktopAccountID = desktopAccountID
		}
		normalizePublishRecordForResponse(&item)
		items = append(items, item)
	}
	return items, nil
}

func publishBatchRequiresCover(accounts []platformAccountSeed) bool {
	for _, account := range accounts {
		if publishPlatformRequiresCover(account.PlatformID) {
			return true
		}
	}
	return false
}

func publishPlatformRequiresCover(platformID string) bool {
	switch strings.TrimSpace(platformID) {
	case "weixin_gzh", "baijiahao", "dongchedi", "smzdm":
		return true
	default:
		return false
	}
}

func pointerStringValue(value *string) string {
	if value == nil {
		return ""
	}
	return *value
}

func normalizePublishStatus(status string) string {
	switch status {
	case "success", "published", "publish_success":
		return "success"
	case "partial_success":
		return "partial_success"
	case "publishing", "pending", "queued", "running":
		return "publishing"
	default:
		return "failed"
	}
}

func newSessionToken() (string, error) {
	buf := make([]byte, 16)
	if _, err := rand.Read(buf); err != nil {
		return "", err
	}
	return hex.EncodeToString(buf), nil
}