YMhut-box-C-/server/unified-management/internal/db/source_store.go

package db

import (
	"database/sql"
	"errors"
)

func (s *Store) UpsertSource(item Source) (Source, error) {
	now := Now()
	if item.SourceID == "" {
		item.SourceID = item.CategoryID + "-" + item.Name
	}
	if item.Method == "" {
		item.Method = "GET"
	}
	item.ProxyMode = normalizeProxyMode(firstNonEmpty(item.ProxyMode, "client_direct"), item.CategoryID, item.Name, item.APIURL)
	if item.URLTemplate == "" {
		item.URLTemplate = item.APIURL
	}
	if item.TimeoutMS <= 0 {
		item.TimeoutMS = 8000
	}
	if item.RetryCount <= 0 {
		item.RetryCount = 1
	}
	if item.CacheSeconds <= 0 {
		item.CacheSeconds = item.CheckIntervalSec
	}
	if item.CacheSeconds <= 0 {
		item.CacheSeconds = 300
	}
	if item.CheckIntervalSec <= 0 {
		item.CheckIntervalSec = item.CacheSeconds
	}
	if item.SupportedFormats == "" {
		item.SupportedFormats = "[]"
	}
	if item.LastStatus == "" {
		item.LastStatus = "unknown"
	}
	if item.CategoryID == "" {
		item.CategoryID = "custom"
	}
	if item.CategoryName == "" {
		item.CategoryName = item.CategoryID
	}
	_, _ = s.exec(`INSERT INTO source_categories (category_id, name, enabled, ui_config, created_at, updated_at)
		VALUES (?, ?, 1, '{}', ?, ?)
		ON CONFLICT (category_id) DO UPDATE SET name = excluded.name, updated_at = excluded.updated_at`,
		item.CategoryID, item.CategoryName, now, now)
	conn, d := s.active()
	query := d.upsert("source_endpoints",
		[]string{"category_id", "category_name", "source_id", "name", "description", "method", "api_url", "url_template", "thumbnail_url", "proxy_mode", "timeout_ms", "retry_count", "cache_seconds", "check_interval_sec", "enabled", "client_visible", "supported_formats", "last_status", "last_latency_ms", "last_checked_at", "last_error", "consecutive_failure", "created_at", "updated_at"},
		[]string{"source_id"})
	if _, err := conn.Exec(d.rebind(query), item.CategoryID, item.CategoryName, item.SourceID, item.Name, item.Description, item.Method, item.APIURL, item.URLTemplate, item.ThumbnailURL,
		item.ProxyMode, item.TimeoutMS, item.RetryCount, item.CacheSeconds, item.CheckIntervalSec, boolInt(item.Enabled), boolInt(item.ClientVisible), item.SupportedFormats,
		item.LastStatus, item.LastLatencyMS, item.LastCheckedAt, item.LastError, item.ConsecutiveFailure, firstNonEmpty(item.CreatedAt, now), now); err != nil {
		s.markFailover(err)
		return Source{}, err
	}
	return s.GetSourceBySourceID(item.SourceID)
}

func (s *Store) GetSourceBySourceID(sourceID string) (Source, error) {
	item, err := scanSourceRow(s.queryRow(sourceSelectSQL()+` WHERE source_id = ?`, sourceID))
	if errors.Is(err, sql.ErrNoRows) {
		return Source{}, errors.New("source not found")
	}
	return item, err
}

func (s *Store) ListSources(includeHidden bool) ([]Source, error) {
	where := ""
	args := []any{}
	if !includeHidden {
		where = " WHERE enabled = 1 AND client_visible = 1"
	}
	rows, err := s.query(sourceSelectSQL()+where+` ORDER BY category_id ASC, name ASC`, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	items := []Source{}
	for rows.Next() {
		item, err := scanSourceRowsCurrent(rows)
		if err != nil {
			return nil, err
		}
		items = append(items, item)
	}
	return items, rows.Err()
}

func (s *Store) CountSources() (int, error) {
	var count int
	err := s.queryRow(`SELECT COUNT(*) FROM source_endpoints`).Scan(&count)
	return count, err
}

func (s *Store) DeleteSource(sourceID string) error {
	_, err := s.exec(`DELETE FROM source_endpoints WHERE source_id = ?`, sourceID)
	return err
}

func (s *Store) RecordSourceCheck(sourceDBID int64, status string, latency int, message string) error {
	now := Now()
	_, err := s.exec(`INSERT INTO endpoint_health_checks (source_db_id, status, latency_ms, error, checked_at) VALUES (?, ?, ?, ?, ?)`,
		sourceDBID, status, latency, sanitize(message), now)
	if err != nil {
		return err
	}
	if status == "ok" {
		_, err = s.exec(`UPDATE source_endpoints SET last_status = ?, last_latency_ms = ?, last_checked_at = ?, last_error = ?, consecutive_failure = 0, updated_at = ? WHERE id = ?`,
			status, latency, now, sanitize(message), now, sourceDBID)
	} else if status == "redirected" {
		_, err = s.exec(`UPDATE source_endpoints SET last_status = ?, last_latency_ms = ?, last_checked_at = ?, last_error = ?, consecutive_failure = 0, updated_at = ? WHERE id = ?`,
			status, latency, now, sanitize(message), now, sourceDBID)
	} else {
		_, err = s.exec(`UPDATE source_endpoints SET last_status = ?, last_latency_ms = ?, last_checked_at = ?, last_error = ?, consecutive_failure = consecutive_failure + 1, updated_at = ? WHERE id = ?`,
			status, latency, now, sanitize(message), now, sourceDBID)
	}
	return err
}

func (s *Store) RecordSourceCall(call SourceCall) error {
	if call.CreatedAt == "" {
		call.CreatedAt = Now()
	}
	_, err := s.exec(`INSERT INTO endpoint_call_logs (source_id, status, latency_ms, error, client, created_at) VALUES (?, ?, ?, ?, ?, ?)`,
		sanitize(call.SourceID), sanitize(call.Status), call.LatencyMS, sanitize(call.Error), sanitize(call.Client), call.CreatedAt)
	return err
}