thefeed/internal/server/server.go

package server

import (
	"bufio"
	"context"
	"fmt"
	"log"
	"os"
	"strings"
	"time"

	"github.com/sartoopjj/thefeed/internal/protocol"
)

// Config holds server configuration.
type Config struct {
	ListenAddr    string
	Domain        string
	Passphrase    string
	ChannelsFile  string
	XAccountsFile string
	XRSSInstances string
	MaxPadding    int
	MsgLimit      int  // max messages per channel (0 = default 15)
	NoTelegram    bool // if true, fetch public channels without Telegram login
	AllowManage   bool // if true, remote channel management and sending via DNS is allowed
	Debug         bool // if true, log every decoded DNS query
	// DNSMediaEnabled toggles the slow DNS-relay path. When false the
	// server still ingests media bytes (so other relays can serve them)
	// but the wire-format DNS flag is unset for clients.
	DNSMediaEnabled     bool
	DNSMediaMaxSize     int64  // per-file cap for the DNS relay (0 = no cap)
	DNSMediaCacheTTL    int    // DNS-relay TTL in minutes
	DNSMediaCompression string // DNS-relay compression: none|gzip|deflate
	FetchInterval       time.Duration // 0 = default 10m; floor enforced by main
	GitHubRelay         GitHubRelayConfig
	Telegram            TelegramConfig
}

// GitHubRelayConfig configures the GitHub fast relay. Active() requires
// Enabled + Token + Repo.
type GitHubRelayConfig struct {
	Enabled    bool
	Token      string
	Repo       string
	Branch     string // default branch to commit to; "" → "main"
	StatePath  string // file used to persist lastSeen across restarts
	MaxBytes   int64
	TTLMinutes int
}

func (g GitHubRelayConfig) Active() bool {
	return g.Enabled && g.Token != "" && g.Repo != ""
}

// Server orchestrates the DNS server and Telegram reader.
type Server struct {
	cfg              Config
	feed             *Feed
	reader           *TelegramReader // nil when --no-telegram
	telegramChannels []string
	xAccounts        []string
}

// New creates a new Server.
func New(cfg Config) (*Server, error) {
	channels, err := loadUsernames(cfg.ChannelsFile)
	if err != nil {
		return nil, fmt.Errorf("load channels: %w", err)
	}
	xAccounts, err := loadUsernames(cfg.XAccountsFile)
	if err != nil {
		return nil, fmt.Errorf("load X accounts: %w", err)
	}

	if len(channels) == 0 && len(xAccounts) == 0 {
		return nil, fmt.Errorf("no channels configured in %s and no X accounts configured in %s", cfg.ChannelsFile, cfg.XAccountsFile)
	}

	log.Printf("[server] loaded %d Telegram channels and %d X accounts", len(channels), len(xAccounts))

	feed := NewFeed(append(append([]string{}, channels...), prefixXAccounts(xAccounts)...))
	return &Server{cfg: cfg, feed: feed, telegramChannels: channels, xAccounts: xAccounts}, nil
}

// Run starts both the DNS server and the Telegram reader.
func (s *Server) Run(ctx context.Context) error {
	queryKey, responseKey, err := protocol.DeriveKeys(s.cfg.Passphrase)
	if err != nil {
		return fmt.Errorf("derive keys: %w", err)
	}

	SetMediaDebugLogs(s.cfg.Debug)

	// Spin up the media cache when at least one relay is enabled. The cache
	// owns the byte pipeline; whether DNS or GitHub serves bytes to clients
	// is controlled by per-relay flags on each MediaMeta.
	anyRelay := s.cfg.DNSMediaEnabled || s.cfg.GitHubRelay.Active()
	if anyRelay {
		ttlMin := s.cfg.DNSMediaCacheTTL
		if ttlMin <= 0 {
			ttlMin = 600
		}
		ttl := time.Duration(ttlMin) * time.Minute
		compName := s.cfg.DNSMediaCompression
		if compName == "" {
			compName = "gzip"
		}
		compression, err := protocol.ParseMediaCompressionName(compName)
		if err != nil {
			return fmt.Errorf("--dns-media-compression: %w", err)
		}
		mediaCache := NewMediaCache(MediaCacheConfig{
			MaxFileBytes:    s.cfg.DNSMediaMaxSize,
			TTL:             ttl,
			Compression:     compression,
			Logf:            logfMedia,
			DNSRelayEnabled: s.cfg.DNSMediaEnabled,
		})
		s.feed.SetMediaCache(mediaCache)
		log.Printf("[server] media: dns=%v max=%d ttl=%s compression=%s",
			s.cfg.DNSMediaEnabled, s.cfg.DNSMediaMaxSize, ttl, compression)
		go s.runMediaSweep(ctx, mediaCache, ttl)

		if s.cfg.GitHubRelay.Active() {
			gh := NewGitHubRelay(s.cfg.GitHubRelay, s.cfg.Domain, s.cfg.Passphrase)
			if gh != nil {
				mediaCache.SetGitHubRelay(gh)
				s.feed.SetGitHubRelay(gh)
				go gh.Run(ctx)
				branch := s.cfg.GitHubRelay.Branch
				if branch == "" {
					branch = "main"
				}
				log.Printf("[server] github relay: repo=%s branch=%s max=%d ttl=%dm",
					gh.Repo(), branch, gh.MaxBytes(), s.cfg.GitHubRelay.TTLMinutes)
			}
		}
	} else {
		log.Println("[server] media disabled (no relays enabled)")
	}

	go startLatestVersionTracker(ctx, s.feed)
	var channelCtl channelRefresher

	// Handle login-only mode
	if s.cfg.Telegram.LoginOnly {
		reader := NewTelegramReader(s.cfg.Telegram, s.telegramChannels, s.feed, 15, 1)
		return reader.Run(ctx)
	}

	// Start Telegram reader in background, or public web fetcher in no-login mode.
	if !s.cfg.NoTelegram {
		msgLimit := s.cfg.MsgLimit
		if msgLimit <= 0 {
			msgLimit = 15
		}
		if len(s.telegramChannels) > 0 {
			reader := NewTelegramReader(s.cfg.Telegram, s.telegramChannels, s.feed, msgLimit, 1)
			reader.SetFetchInterval(s.cfg.FetchInterval)
			s.reader = reader
			channelCtl = reader
			go func() {
				log.Println("[telegram] reader goroutine started")
				if err := reader.Run(ctx); err != nil && ctx.Err() == nil {
					log.Printf("[telegram] reader goroutine STOPPED with error: %v", err)
				} else {
					log.Println("[telegram] reader goroutine exited")
				}
			}()
		} else {
			s.feed.SetTelegramLoggedIn(true)
		}
	} else {
		msgLimit := s.cfg.MsgLimit
		if msgLimit <= 0 {
			msgLimit = 15
		}
		publicReader := NewPublicReader(s.telegramChannels, s.feed, msgLimit, 1)
		publicReader.SetFetchInterval(s.cfg.FetchInterval)
		channelCtl = publicReader
		go func() {
			log.Println("[public] reader goroutine started")
			if err := publicReader.Run(ctx); err != nil && ctx.Err() == nil {
				log.Printf("[public] reader goroutine STOPPED with error: %v", err)
			} else {
				log.Println("[public] reader goroutine exited")
			}
		}()
		log.Println("[server] running without Telegram login; fetching public channels via t.me")
	}

	var xReader *XPublicReader
	if len(s.xAccounts) > 0 {
		msgLimit := s.cfg.MsgLimit
		if msgLimit <= 0 {
			msgLimit = 15
		}
		xReader = NewXPublicReader(s.xAccounts, s.feed, msgLimit, len(s.telegramChannels)+1, s.cfg.XRSSInstances)
		xReader.SetFetchInterval(s.cfg.FetchInterval)
		go func() {
			log.Println("[x] reader goroutine started")
			if err := xReader.Run(ctx); err != nil && ctx.Err() == nil {
				log.Printf("[x] reader goroutine STOPPED with error: %v", err)
			} else {
				log.Println("[x] reader goroutine exited")
			}
		}()
		log.Printf("[server] enabled X source for %d accounts", len(s.xAccounts))
	}

	// Start DNS server (blocking, respects ctx cancellation)
	maxPad := s.cfg.MaxPadding
	if maxPad == 0 {
		maxPad = protocol.DefaultMaxPadding
	}
	dnsServer := NewDNSServer(s.cfg.ListenAddr, s.cfg.Domain, s.feed, queryKey, responseKey, maxPad, s.reader, s.cfg.AllowManage, s.cfg.ChannelsFile, s.xAccounts, s.cfg.Debug)
	if channelCtl != nil {
		dnsServer.SetChannelRefresher(channelCtl)
	}
	if xReader != nil {
		dnsServer.AddRefresher(xReader)
		dnsServer.SetXReader(xReader)
	}
	return dnsServer.ListenAndServe(ctx)
}

func loadUsernames(path string) ([]string, error) {
	f, err := os.Open(path)
	if err != nil {
		if os.IsNotExist(err) {
			return []string{}, nil
		}
		return nil, err
	}
	defer func() {
		if err := f.Close(); err != nil {
			log.Printf("[server] close usernames file: %v", err)
		}
	}()

	var users []string
	scanner := bufio.NewScanner(f)
	for scanner.Scan() {
		line := strings.TrimSpace(scanner.Text())
		if line == "" || strings.HasPrefix(line, "#") {
			continue
		}
		name := strings.TrimPrefix(line, "@")
		users = append(users, name)
	}
	return users, scanner.Err()
}

func prefixXAccounts(accounts []string) []string {
	out := make([]string, len(accounts))
	for i, a := range accounts {
		out[i] = "x/" + a
	}
	return out
}

// runMediaSweep periodically evicts expired entries from the cache. The
// interval is min(ttl/4, 5min) so we don't waste cycles on long-TTL configs
// while still reclaiming slots in time under steady-state churn.
func (s *Server) runMediaSweep(ctx context.Context, cache *MediaCache, ttl time.Duration) {
	if cache == nil {
		return
	}
	interval := ttl / 4
	if interval <= 0 || interval > 5*time.Minute {
		interval = 5 * time.Minute
	}
	if interval < 30*time.Second {
		interval = 30 * time.Second
	}
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			cache.Sweep()
		}
	}
}