From 8ec1f5e68689a020b163f7c127eb1c67ef0f9ae6 Mon Sep 17 00:00:00 2001 From: moeid8161-boop Date: Mon, 22 Jun 2026 05:25:52 +0330 Subject: [PATCH] feat: support separate dns audio limit, channel-specific overrides, and optional opus transcoding - Add --dns-audio-max-size flag and configuration parsing to allow a distinct download limit exclusively for audio/voice files. - Add support for channel-specific overrides for both dns-media-max-size and dns-audio-max-size inside channels config. - Add optional server-side audio-to-opus transcode capability, gated behind the THEFEED_OPUS_TRANSCODE=1 environment variable. - Fix: Preserve the case of private channel invite hashes in limits parsing. --- cmd/server/main.go | 7 + internal/protocol/media.go | 13 +- internal/protocol/media_test.go | 19 +-- internal/server/dns.go | 5 +- internal/server/media.go | 180 +++++++++++++++++++-- internal/server/media_http.go | 12 +- internal/server/media_telegram.go | 37 +++-- internal/server/media_test.go | 203 ++++++++++++++++++++++++ internal/server/private_channels.go | 6 +- internal/server/profile_pic_telegram.go | 2 +- internal/server/public.go | 20 ++- internal/server/server.go | 175 +++++++++++++++++++- internal/server/telegram.go | 17 +- internal/server/xpublic.go | 10 +- 14 files changed, 645 insertions(+), 61 deletions(-) diff --git a/cmd/server/main.go b/cmd/server/main.go index 9c18caf..11ae8df 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -44,6 +44,7 @@ func main() { debug := flag.Bool("debug", false, "Log every decoded DNS query") dnsMediaEnabled := flag.Bool("dns-media-enabled", false, "Serve media via DNS (slow relay)") dnsMediaMaxSizeKB := flag.Int("dns-media-max-size", 100, "Per-file cap for the DNS relay in KB (0 = no cap)") + dnsAudioMaxSizeKB := flag.Int("dns-audio-max-size", 0, "Per-file cap for audio/voice files in the DNS relay in KB (0 = fallback to dns-media-max-size)") dnsMediaCacheTTLMin := flag.Int("dns-media-cache-ttl", 600, "TTL for DNS-relay cached media, in minutes") dnsMediaCompression := flag.String("dns-media-compression", "gzip", "Compression for DNS-relay media bytes: none|gzip|deflate") chatDomains := flag.String("chat-domains", "", "Comma-separated dedicated sub-domains for the chat messenger (empty = chat off)") @@ -327,6 +328,11 @@ func main() { *dnsMediaMaxSizeKB = n } } + if env := os.Getenv("THEFEED_DNS_AUDIO_MAX_SIZE_KB"); env != "" { + if n, err := strconv.Atoi(env); err == nil { + *dnsAudioMaxSizeKB = n + } + } if env := os.Getenv("THEFEED_DNS_MEDIA_CACHE_TTL_MIN"); env != "" { if n, err := strconv.Atoi(env); err == nil { *dnsMediaCacheTTLMin = n @@ -377,6 +383,7 @@ func main() { Debug: *debug, DNSMediaEnabled: *dnsMediaEnabled, DNSMediaMaxSize: int64(*dnsMediaMaxSizeKB) * 1024, + DNSAudioMaxSize: int64(*dnsAudioMaxSizeKB) * 1024, DNSMediaCacheTTL: *dnsMediaCacheTTLMin, DNSMediaCompression: *dnsMediaCompression, FetchInterval: time.Duration(*fetchIntervalMin) * time.Minute, diff --git a/internal/protocol/media.go b/internal/protocol/media.go index 6f7e54f..d0bbf03 100644 --- a/internal/protocol/media.go +++ b/internal/protocol/media.go @@ -6,6 +6,7 @@ import ( "hash/fnv" "strconv" "strings" + "unicode" ) // Relay indices: each MediaMeta.Relays[N] flags whether the file is @@ -147,11 +148,13 @@ func SanitiseMediaFilename(s string) string { func filterFilenameRunes(s string) string { var b strings.Builder for _, r := range s { - switch { - case r >= '0' && r <= '9', - r >= 'A' && r <= 'Z', - r >= 'a' && r <= 'z', - r == '.', r == '_', r == '-': + if r == unicode.ReplacementChar { + continue + } + if r == ':' || r == '/' || r == '\\' || r == '*' || r == '?' || r == '"' || r == '<' || r == '>' || r == '|' { + continue + } + if unicode.IsPrint(r) { b.WriteRune(r) } } diff --git a/internal/protocol/media_test.go b/internal/protocol/media_test.go index 6cf13e6..d5666f0 100644 --- a/internal/protocol/media_test.go +++ b/internal/protocol/media_test.go @@ -132,15 +132,16 @@ func TestParseMediaTextUnknownRelaysIgnored(t *testing.T) { func TestSanitiseMediaFilename(t *testing.T) { cases := map[string]string{ - "": "", - "report.zip": "report.zip", - "path/to/report.zip": "report.zip", - "..": "", - "a:b\nc.txt": "abc.txt", - "hello": "hello", - "WeIrD-Name_v2.tar.gz": "WeIrD-Name_v2.tar.gz", - "\xff\xfe.txt": "media.txt", - "\u062d\u0645\u0644\u0647.zip": "media.zip", + "": "", + "report.zip": "report.zip", + "path/to/report.zip": "report.zip", + "..": "", + "a:b\nc.txt": "abc.txt", + "hello": "hello", + "WeIrD-Name_v2.tar.gz": "WeIrD-Name_v2.tar.gz", + "\xff\xfe.txt": "media.txt", + "\u062d\u0645\u0644\u0647.zip": "حمله.zip", + "یه نام تست.mp3": "یه نام تست.mp3", } for in, want := range cases { if got := SanitiseMediaFilename(in); got != want { diff --git a/internal/server/dns.go b/internal/server/dns.go index fadb987..f998130 100644 --- a/internal/server/dns.go +++ b/internal/server/dns.go @@ -828,7 +828,10 @@ func loadChannelsFromFile(path string) ([]string, error) { if line == "" || strings.HasPrefix(line, "#") { continue } - channels = append(channels, strings.TrimPrefix(line, "@")) + parts := strings.Fields(line) + if len(parts) > 0 { + channels = append(channels, strings.TrimPrefix(parts[0], "@")) + } } return channels, scanner.Err() } diff --git a/internal/server/media.go b/internal/server/media.go index 4af0d34..88152c3 100644 --- a/internal/server/media.go +++ b/internal/server/media.go @@ -9,6 +9,10 @@ import ( "fmt" "hash/crc32" "io" + "os" + "os/exec" + "path/filepath" + "strings" "sync" "sync/atomic" "time" @@ -25,10 +29,11 @@ import ( // The cache is safe for concurrent use. Hot-path operations (Store, GetBlock) // are O(log n) at worst and typically O(1) with the help of two side maps. type MediaCache struct { - maxFileBytes int64 - ttl time.Duration - compression protocol.MediaCompression - dnsEnabled bool // when false, RelayDNS stays unset on the wire + maxFileBytes int64 + maxAudioBytes int64 + ttl time.Duration + compression protocol.MediaCompression + dnsEnabled bool // when false, RelayDNS stays unset on the wire logf func(format string, args ...interface{}) @@ -70,6 +75,7 @@ type mediaEntry struct { // MediaCacheConfig configures a new MediaCache. type MediaCacheConfig struct { MaxFileBytes int64 + MaxAudioBytes int64 TTL time.Duration Compression protocol.MediaCompression Logf func(format string, args ...interface{}) @@ -93,11 +99,12 @@ func NewMediaCache(cfg MediaCacheConfig) *MediaCache { logf = func(string, ...interface{}) {} } return &MediaCache{ - maxFileBytes: cfg.MaxFileBytes, - ttl: cfg.TTL, - compression: cfg.Compression, - dnsEnabled: cfg.DNSRelayEnabled, - logf: logf, + maxFileBytes: cfg.MaxFileBytes, + maxAudioBytes: cfg.MaxAudioBytes, + ttl: cfg.TTL, + compression: cfg.Compression, + dnsEnabled: cfg.DNSRelayEnabled, + logf: logf, byKey: make(map[string]*mediaEntry), byChannel: make(map[uint16]*mediaEntry), byHash: make(map[uint32]*mediaEntry), @@ -127,7 +134,9 @@ func (c *MediaCache) Store(cacheKey, tag string, content []byte, mimeType, filen // upload. SkipGitHub keeps the DNS allocation but skips the upload — // used when many small siblings share one bundled GitHub upload. type MediaCacheStoreOptions struct { - SkipGitHub bool + SkipGitHub bool + MaxFileBytesOverride *int64 + MaxAudioBytesOverride *int64 } // StoreWithOptions is Store with selective relay control. @@ -138,8 +147,53 @@ func (c *MediaCache) StoreWithOptions(cacheKey, tag string, content []byte, mime if tag == "" { tag = protocol.MediaFile } + if tag == protocol.MediaAudio && os.Getenv("THEFEED_OPUS_TRANSCODE") == "1" { + if transcoded, err := c.transcodeToOpus(content); err == nil { + content = transcoded + mimeType = "audio/ogg" + filename = changeExtensionToOpus(filename) + } else { + c.logf("media: ffmpeg transcoding failed, using original: %v", err) + } + } size := int64(len(content)) - if max := c.MaxAcceptableBytes(); max > 0 && size > max { + + dnsLimit := c.maxFileBytes + if opts.MaxFileBytesOverride != nil && *opts.MaxFileBytesOverride != -1 { + dnsLimit = *opts.MaxFileBytesOverride + } + audioLimit := c.maxAudioBytes + if opts.MaxAudioBytesOverride != nil && *opts.MaxAudioBytesOverride != -1 { + audioLimit = *opts.MaxAudioBytesOverride + } + + dns := dnsLimit + if audioLimit > 0 && isAudioOrVoice(tag, mimeType) { + dns = audioLimit + } + + var ghMax int64 + c.mu.RLock() + gh := c.gh + c.mu.RUnlock() + if gh != nil { + ghMax = gh.MaxBytes() + } + + var max int64 + if (dns == 0 && c.dnsEnabled) || (gh != nil && ghMax == 0) { + max = 0 + } else if !c.dnsEnabled { + max = ghMax + } else if gh == nil { + max = dns + } else if ghMax > dns { + max = ghMax + } else { + max = dns + } + + if max > 0 && size > max { atomic.AddUint64(&c.storeRejected, 1) return protocol.MediaMeta{ Tag: tag, @@ -147,7 +201,11 @@ func (c *MediaCache) StoreWithOptions(cacheKey, tag string, content []byte, mime Relays: nil, }, ErrTooLarge } - dnsFits := c.maxFileBytes == 0 || size <= c.maxFileBytes + + if audioLimit > 0 && isAudioOrVoice(tag, mimeType) { + dnsLimit = audioLimit + } + dnsFits := dnsLimit == 0 || size <= dnsLimit now := time.Now() hash := crc32.ChecksumIEEE(content) @@ -538,10 +596,14 @@ func (c *MediaCache) TouchRelayEntries() { } } -// MaxAcceptableBytes returns the largest file size any enabled relay would -// accept. Callers use it as the "should we even fetch this?" gate so that -// files which fit GitHub but not DNS still get pulled. 0 means "no cap". -func (c *MediaCache) MaxAcceptableBytes() int64 { +// isAudioOrVoice returns true if the given tag or mimeType indicates an audio or voice file. +func isAudioOrVoice(tag, mimeType string) bool { + return tag == protocol.MediaAudio || strings.HasPrefix(strings.ToLower(mimeType), "audio/") +} + +// MaxAcceptableBytesFor returns the largest file size any enabled relay would +// accept for a specific tag and mimeType. 0 means "no cap". +func (c *MediaCache) MaxAcceptableBytesFor(tag, mimeType string) int64 { if c == nil { return 0 } @@ -549,6 +611,9 @@ func (c *MediaCache) MaxAcceptableBytes() int64 { gh := c.gh c.mu.RUnlock() dns := c.maxFileBytes + if c.maxAudioBytes > 0 && isAudioOrVoice(tag, mimeType) { + dns = c.maxAudioBytes + } var ghMax int64 if gh != nil { ghMax = gh.MaxBytes() @@ -569,6 +634,13 @@ func (c *MediaCache) MaxAcceptableBytes() int64 { return dns } +// MaxAcceptableBytes returns the largest file size any enabled relay would +// accept. Callers use it as the "should we even fetch this?" gate so that +// files which fit GitHub but not DNS still get pulled. 0 means "no cap". +func (c *MediaCache) MaxAcceptableBytes() int64 { + return c.MaxAcceptableBytesFor("", "") +} + // splitMediaBlocks compresses the content (when compression != none), // prepends the protocol media header, then splits the result into // randomly-sized blocks. The CRC32 carried in the header is over the @@ -641,3 +713,79 @@ func DecompressMediaBytes(r io.Reader, compression protocol.MediaCompression) (i } return nil, fmt.Errorf("unsupported media compression: %d", compression) } + +func (c *MediaCache) transcodeToOpus(content []byte) ([]byte, error) { + ffmpegPath, err := exec.LookPath("ffmpeg") + if err != nil { + return nil, fmt.Errorf("ffmpeg not found: %w", err) + } + + // Write original audio to a temp file. + inTemp, err := os.CreateTemp("", "feed_audio_in_*") + if err != nil { + return nil, fmt.Errorf("failed to create input temp file: %w", err) + } + inPath := inTemp.Name() + + if _, err := inTemp.Write(content); err != nil { + inTemp.Close() + os.Remove(inPath) + return nil, fmt.Errorf("failed to write input temp file: %w", err) + } + inTemp.Close() + // content slice is no longer needed by this function — the caller + // will replace it with the transcoded bytes, allowing GC to reclaim + // the original audio memory. + + // Prepare output temp file path. + outTemp, err := os.CreateTemp("", "feed_audio_out_*.opus") + if err != nil { + os.Remove(inPath) + return nil, fmt.Errorf("failed to create output temp file: %w", err) + } + outPath := outTemp.Name() + outTemp.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + + // Try libopus first (higher quality), fall back to native opus encoder. + cmd := exec.CommandContext(ctx, ffmpegPath, "-y", "-i", inPath, "-vn", "-c:a", "libopus", "-b:a", "16k", "-ac", "1", "-ar", "16000", "-vbr", "on", outPath) + output, err := cmd.CombinedOutput() + if err != nil { + cmdFallback := exec.CommandContext(ctx, ffmpegPath, "-y", "-i", inPath, "-vn", "-c:a", "opus", "-b:a", "16k", "-ac", "1", "-ar", "16000", "-vbr", "on", outPath) + outputFallback, errFallback := cmdFallback.CombinedOutput() + if errFallback != nil { + os.Remove(inPath) + os.Remove(outPath) + return nil, fmt.Errorf("ffmpeg transcoding failed: %v (libopus output: %s) (opus output: %s)", errFallback, string(output), string(outputFallback)) + } + } + + // Eagerly remove input file — frees disk space immediately. + os.Remove(inPath) + + opusBytes, err := os.ReadFile(outPath) + // Eagerly remove output file — frees disk space immediately. + os.Remove(outPath) + if err != nil { + return nil, fmt.Errorf("failed to read output temp file: %w", err) + } + + if len(opusBytes) == 0 { + return nil, fmt.Errorf("transcoded file is empty") + } + + return opusBytes, nil +} + +func changeExtensionToOpus(filename string) string { + if filename == "" { + return "audio.opus" + } + ext := filepath.Ext(filename) + if ext == "" { + return filename + ".opus" + } + return strings.TrimSuffix(filename, ext) + ".opus" +} diff --git a/internal/server/media_http.go b/internal/server/media_http.go index 392f323..724c994 100644 --- a/internal/server/media_http.go +++ b/internal/server/media_http.go @@ -90,7 +90,10 @@ func downloadHTTPMedia(ctx context.Context, cache *MediaCache, tag, rawURL strin return protocol.MediaMeta{}, false } - maxBytes := cache.MaxAcceptableBytes() + maxBytes := cache.MaxAcceptableBytesFor(tag, ctype) + if overrideMax, ok := GetMaxBytesFromContext(ctx, tag, ctype, cache); ok { + maxBytes = overrideMax + } if maxBytes > 0 && resp.ContentLength > 0 && resp.ContentLength > maxBytes { size := resp.ContentLength return protocol.MediaMeta{ @@ -121,7 +124,12 @@ func downloadHTTPMedia(ctx context.Context, cache *MediaCache, tag, rawURL strin }, true } - meta, err := cache.Store(cacheKey, tag, bytes, resp.Header.Get("Content-Type"), urlBaseName(parsed)) + var storeOpts MediaCacheStoreOptions + if maxFile, maxAudio, ok := GetContextLimits(ctx); ok { + storeOpts.MaxFileBytesOverride = &maxFile + storeOpts.MaxAudioBytesOverride = &maxAudio + } + meta, err := cache.StoreWithOptions(cacheKey, tag, bytes, resp.Header.Get("Content-Type"), urlBaseName(parsed), storeOpts) if err != nil { if errors.Is(err, ErrTooLarge) { return meta, true diff --git a/internal/server/media_telegram.go b/internal/server/media_telegram.go index f52e6de..0178371 100644 --- a/internal/server/media_telegram.go +++ b/internal/server/media_telegram.go @@ -80,7 +80,11 @@ func (tr *TelegramReader) downloadTelegramPhoto(ctx context.Context, api *tg.Cli // Honour the configured max-size early so we don't even open the RPC for // objects no enabled relay would accept. Files that fit GitHub but not // DNS still get fetched. - if maxBytes := cache.MaxAcceptableBytes(); maxBytes > 0 && bestBytes > maxBytes { + maxBytes := cache.MaxAcceptableBytesFor(protocol.MediaImage, "image/jpeg") + if overrideMax, ok := GetMaxBytesFromContext(ctx, protocol.MediaImage, "image/jpeg", cache); ok { + maxBytes = overrideMax + } + if maxBytes > 0 && bestBytes > maxBytes { return protocol.MediaMeta{ Tag: protocol.MediaImage, Size: bestBytes, @@ -94,7 +98,7 @@ func (tr *TelegramReader) downloadTelegramPhoto(ctx context.Context, api *tg.Cli FileReference: photo.FileReference, ThumbSize: bestType, } - bytes, err := tr.downloadTelegramFile(ctx, api, loc, bestBytes) + bytes, err := tr.downloadTelegramFile(ctx, api, loc, bestBytes, maxBytes) if err != nil { // Transient fetch error (network, FILE_REFERENCE_EXPIRED, etc.). // We don't mark the message as non-downloadable in that case — @@ -105,7 +109,12 @@ func (tr *TelegramReader) downloadTelegramPhoto(ctx context.Context, api *tg.Cli return protocol.MediaMeta{}, false } - meta, err := cache.Store(cacheKey, protocol.MediaImage, bytes, "image/jpeg", "") + var storeOpts MediaCacheStoreOptions + if maxFile, maxAudio, ok := GetContextLimits(ctx); ok { + storeOpts.MaxFileBytesOverride = &maxFile + storeOpts.MaxAudioBytesOverride = &maxAudio + } + meta, err := cache.StoreWithOptions(cacheKey, protocol.MediaImage, bytes, "image/jpeg", "", storeOpts) if err != nil { // ErrTooLarge is reported as non-downloadable; any other store error // is just dropped to legacy. @@ -129,7 +138,11 @@ func (tr *TelegramReader) downloadTelegramDocument(ctx context.Context, api *tg. return protocol.MediaMeta{}, false } - if maxBytes := cache.MaxAcceptableBytes(); maxBytes > 0 && doc.Size > maxBytes { + maxBytes := cache.MaxAcceptableBytesFor(tag, doc.MimeType) + if overrideMax, ok := GetMaxBytesFromContext(ctx, tag, doc.MimeType, cache); ok { + maxBytes = overrideMax + } + if maxBytes > 0 && doc.Size > maxBytes { return protocol.MediaMeta{ Tag: tag, Size: doc.Size, @@ -143,7 +156,7 @@ func (tr *TelegramReader) downloadTelegramDocument(ctx context.Context, api *tg. FileReference: doc.FileReference, ThumbSize: "", } - bytes, err := tr.downloadTelegramFile(ctx, api, loc, doc.Size) + bytes, err := tr.downloadTelegramFile(ctx, api, loc, doc.Size, maxBytes) if err != nil { // See note in downloadTelegramPhoto: transient fetch errors should // not be surfaced as "non-downloadable", they should fall through @@ -152,7 +165,12 @@ func (tr *TelegramReader) downloadTelegramDocument(ctx context.Context, api *tg. return protocol.MediaMeta{}, false } - meta, err := cache.Store(cacheKey, tag, bytes, doc.MimeType, filename) + var storeOpts MediaCacheStoreOptions + if maxFile, maxAudio, ok := GetContextLimits(ctx); ok { + storeOpts.MaxFileBytesOverride = &maxFile + storeOpts.MaxAudioBytesOverride = &maxAudio + } + meta, err := cache.StoreWithOptions(cacheKey, tag, bytes, doc.MimeType, filename, storeOpts) if err != nil { if errors.Is(err, ErrTooLarge) { return meta, true @@ -167,12 +185,7 @@ func (tr *TelegramReader) downloadTelegramDocument(ctx context.Context, api *tg. // when expectedSize <= 0) from the given Telegram file location. It enforces // the configured max-size cap defensively so a file that lies about its size // still can't blow past the limit on the wire. -func (tr *TelegramReader) downloadTelegramFile(ctx context.Context, api *tg.Client, loc tg.InputFileLocationClass, expectedSize int64) ([]byte, error) { - cache := tr.feed.MediaCache() - maxBytes := int64(0) - if cache != nil { - maxBytes = cache.MaxAcceptableBytes() - } +func (tr *TelegramReader) downloadTelegramFile(ctx context.Context, api *tg.Client, loc tg.InputFileLocationClass, expectedSize int64, maxBytes int64) ([]byte, error) { var ( out []byte diff --git a/internal/server/media_test.go b/internal/server/media_test.go index 0eaad3e..1c0c195 100644 --- a/internal/server/media_test.go +++ b/internal/server/media_test.go @@ -2,8 +2,12 @@ package server import ( "bytes" + "context" "errors" "hash/crc32" + "os" + "os/exec" + "path/filepath" "strings" "testing" "time" @@ -323,3 +327,202 @@ func TestMediaCacheMetadataRoundTrip(t *testing.T) { t.Fatalf("caption = %q", caption) } } + +func TestMediaCacheAudioTranscode(t *testing.T) { + t.Setenv("THEFEED_OPUS_TRANSCODE", "1") + // 1. Test fallback with invalid/garbage bytes + cache := newTestCache(0, time.Hour) + garbage := []byte("not-real-audio-data-at-all") + meta, err := cache.Store("garbage-audio", protocol.MediaAudio, garbage, "audio/mp3", "test.mp3") + if err != nil { + t.Fatalf("Store garbage audio: %v", err) + } + // It should have fallen back to original bytes and filename + if meta.Filename != "test.mp3" { + t.Errorf("expected fallback filename 'test.mp3', got %q", meta.Filename) + } + if meta.Size != int64(len(garbage)) { + t.Errorf("expected fallback size %d, got %d", len(garbage), meta.Size) + } + + // 2. Test successful transcoding with a real valid audio file (if ffmpeg is available) + _, err = exec.LookPath("ffmpeg") + if err != nil { + t.Skip("skipping transcode test; ffmpeg not found in PATH") + } + + // Generate a tiny mp3 file using ffmpeg + tmpMP3, err := os.CreateTemp("", "test_sine_*.mp3") + if err != nil { + t.Fatalf("failed to create temp mp3: %v", err) + } + tmpMP3Path := tmpMP3.Name() + tmpMP3.Close() + defer os.Remove(tmpMP3Path) + + cmd := exec.Command("ffmpeg", "-y", "-f", "lavfi", "-i", "sine=frequency=1000:duration=1", "-acodec", "libmp3lame", tmpMP3Path) + if out, err := cmd.CombinedOutput(); err != nil { + // If libmp3lame is not available, try generating wav instead + t.Logf("mp3 generation failed, trying wav: %v (output: %s)", err, string(out)) + tmpWav := strings.TrimSuffix(tmpMP3Path, ".mp3") + ".wav" + cmdWav := exec.Command("ffmpeg", "-y", "-f", "lavfi", "-i", "sine=frequency=1000:duration=1", tmpWav) + if outWav, errWav := cmdWav.CombinedOutput(); errWav != nil { + t.Skipf("skipping transcode test; cannot generate source audio: %v (output: %s)", errWav, string(outWav)) + } + tmpMP3Path = tmpWav + defer os.Remove(tmpWav) + } + + audioBytes, err := os.ReadFile(tmpMP3Path) + if err != nil { + t.Fatalf("failed to read generated audio: %v", err) + } + + // Store the real audio + origFilename := filepath.Base(tmpMP3Path) + metaAudio, err := cache.Store("real-audio", protocol.MediaAudio, audioBytes, "audio/mpeg", origFilename) + if err != nil { + t.Fatalf("Store real audio: %v", err) + } + + // It should have transcoded to opus + expectedFilename := changeExtensionToOpus(origFilename) + if metaAudio.Filename != expectedFilename { + t.Errorf("expected transcoded filename %q, got %q", expectedFilename, metaAudio.Filename) + } + if metaAudio.Size == int64(len(audioBytes)) { + t.Errorf("expected different size after transcoding, but got same size %d", metaAudio.Size) + } + // Verify that the retrieved data is not equal to original but exists + block, err := cache.GetBlock(metaAudio.Channel, 0) + if err != nil { + t.Fatalf("failed to get block: %v", err) + } + if len(block) == 0 { + t.Fatalf("retrieved block is empty") + } +} + +func TestMediaAudioMaxSize(t *testing.T) { + // 1. Fallback behavior (MaxAudioBytes = 0) + cacheFallback := NewMediaCache(MediaCacheConfig{ + MaxFileBytes: 100, + MaxAudioBytes: 0, + TTL: time.Hour, + DNSRelayEnabled: true, + }) + + // Non-audio file within limit + _, err := cacheFallback.Store("img-small", protocol.MediaImage, make([]byte, 50), "image/jpeg", "test.jpg") + if err != nil { + t.Errorf("expected success for small image, got error: %v", err) + } + + // Non-audio file exceeding limit + _, err = cacheFallback.Store("img-large", protocol.MediaImage, make([]byte, 150), "image/jpeg", "test.jpg") + if err == nil { + t.Errorf("expected error for large image exceeding MaxFileBytes, got nil") + } + + // Audio file within limit + _, err = cacheFallback.Store("audio-small", protocol.MediaAudio, make([]byte, 50), "audio/mpeg", "test.mp3") + if err != nil { + t.Errorf("expected success for small audio, got error: %v", err) + } + + // Audio file exceeding limit + _, err = cacheFallback.Store("audio-large", protocol.MediaAudio, make([]byte, 150), "audio/mpeg", "test.mp3") + if err == nil { + t.Errorf("expected error for large audio exceeding MaxFileBytes under fallback, got nil") + } + + // 2. Specific audio size limit behavior (MaxAudioBytes = 200, MaxFileBytes = 100) + cacheCustom := NewMediaCache(MediaCacheConfig{ + MaxFileBytes: 100, + MaxAudioBytes: 200, + TTL: time.Hour, + DNSRelayEnabled: true, + }) + + // Non-audio file within MaxFileBytes (100) + _, err = cacheCustom.Store("img-custom-ok", protocol.MediaImage, make([]byte, 80), "image/jpeg", "test.jpg") + if err != nil { + t.Errorf("expected success for image under MaxFileBytes, got error: %v", err) + } + + // Non-audio file exceeding MaxFileBytes (100) but under MaxAudioBytes (200) + _, err = cacheCustom.Store("img-custom-too-large", protocol.MediaImage, make([]byte, 150), "image/jpeg", "test.jpg") + if err == nil { + t.Errorf("expected error for image exceeding MaxFileBytes, got nil") + } + + // Audio file exceeding MaxFileBytes (100) but under MaxAudioBytes (200) + _, err = cacheCustom.Store("audio-custom-ok", protocol.MediaAudio, make([]byte, 150), "audio/mpeg", "test.mp3") + if err != nil { + t.Errorf("expected success for audio exceeding MaxFileBytes but under MaxAudioBytes, got error: %v", err) + } + + // Audio file by MIME type exceeding MaxFileBytes (100) but under MaxAudioBytes (200) + _, err = cacheCustom.Store("audio-mime-ok", protocol.MediaFile, make([]byte, 150), "audio/ogg", "test.ogg") + if err != nil { + t.Errorf("expected success for file with audio MIME type exceeding MaxFileBytes but under MaxAudioBytes, got error: %v", err) + } + + // Audio file exceeding MaxAudioBytes (200) + _, err = cacheCustom.Store("audio-custom-too-large", protocol.MediaAudio, make([]byte, 250), "audio/mpeg", "test.mp3") + if err == nil { + t.Errorf("expected error for audio exceeding MaxAudioBytes, got nil") + } +} + +func TestMediaCacheOverrides(t *testing.T) { + cache := NewMediaCache(MediaCacheConfig{ + MaxFileBytes: 100, + MaxAudioBytes: 200, + TTL: time.Hour, + DNSRelayEnabled: true, + }) + + // 1. Check default context limits (none) + ctx := context.Background() + if _, _, ok := GetContextLimits(ctx); ok { + t.Error("expected ok=false for empty context") + } + + // 2. Check context limit propagation + ctx = WithContextLimits(ctx, 50, 60) + maxF, maxA, ok := GetContextLimits(ctx) + if !ok || maxF != 50 || maxA != 60 { + t.Errorf("expected limits (50, 60), got (%d, %d), ok=%v", maxF, maxA, ok) + } + + // 3. Check GetMaxBytesFromContext override calculation + overrideMax, ok := GetMaxBytesFromContext(ctx, protocol.MediaImage, "image/jpeg", cache) + if !ok || overrideMax != 50 { + t.Errorf("expected max bytes 50 for image, got %d, ok=%v", overrideMax, ok) + } + overrideMax, ok = GetMaxBytesFromContext(ctx, protocol.MediaAudio, "audio/mpeg", cache) + if !ok || overrideMax != 60 { + t.Errorf("expected max bytes 60 for audio, got %d, ok=%v", overrideMax, ok) + } + + // 4. Test StoreWithOptions respecting overrides + var storeOpts MediaCacheStoreOptions + fLimit, aLimit := int64(300), int64(400) + storeOpts.MaxFileBytesOverride = &fLimit + storeOpts.MaxAudioBytesOverride = &aLimit + + // Without override, image of size 150 would be rejected by cache (global limit 100) + _, err := cache.StoreWithOptions("k-overlimit-no-override", protocol.MediaImage, make([]byte, 150), "image/jpeg", "", MediaCacheStoreOptions{}) + if err == nil { + t.Error("expected error for image exceeding global limit without override") + } + + // With override (300), image of size 150 should succeed + _, err = cache.StoreWithOptions("k-overlimit-override", protocol.MediaImage, make([]byte, 150), "image/jpeg", "", storeOpts) + if err != nil { + t.Errorf("expected success for image with override limit 300, got error: %v", err) + } +} + + diff --git a/internal/server/private_channels.go b/internal/server/private_channels.go index b0fd0f3..661cecc 100644 --- a/internal/server/private_channels.go +++ b/internal/server/private_channels.go @@ -119,7 +119,11 @@ func LoadPrivateInvites(path string) ([]string, error) { if line == "" || strings.HasPrefix(line, "#") { continue } - hash, err := ParseInviteHash(line) + parts := strings.Fields(line) + if len(parts) == 0 { + continue + } + hash, err := ParseInviteHash(parts[0]) if err != nil { log.Printf("[server] private_channels.txt:%d: %v", lineNum, err) continue diff --git a/internal/server/profile_pic_telegram.go b/internal/server/profile_pic_telegram.go index f8780a4..433604a 100644 --- a/internal/server/profile_pic_telegram.go +++ b/internal/server/profile_pic_telegram.go @@ -38,7 +38,7 @@ func (tr *TelegramReader) fetchProfilePhoto(ctx context.Context, api *tg.Client, PhotoID: photoID, } loc.Big = false - return tr.downloadTelegramFile(ctx, api, loc, 0) + return tr.downloadTelegramFile(ctx, api, loc, 0, 0) } // fetchAllProfilePhotos downloads avatars (public by username, diff --git a/internal/server/public.go b/internal/server/public.go index 711e7a5..63459fc 100644 --- a/internal/server/public.go +++ b/internal/server/public.go @@ -34,6 +34,7 @@ type PublicReader struct { fetchInterval time.Duration refreshCh chan struct{} + limits map[string]ChannelLimits } // SetFetchInterval overrides the default 10m fetch cadence. Caller must @@ -49,7 +50,7 @@ func (pr *PublicReader) SetFetchInterval(d time.Duration) { } // NewPublicReader creates a reader for public channels without Telegram login. -func NewPublicReader(channelUsernames []string, feed *Feed, msgLimit int, baseCh int) *PublicReader { +func NewPublicReader(channelUsernames []string, feed *Feed, msgLimit int, baseCh int, limits map[string]ChannelLimits) *PublicReader { cleaned := make([]string, len(channelUsernames)) for i, u := range channelUsernames { cleaned[i] = strings.TrimPrefix(strings.TrimSpace(u), "@") @@ -61,18 +62,19 @@ func NewPublicReader(channelUsernames []string, feed *Feed, msgLimit int, baseCh baseCh = 1 } return &PublicReader{ - channels: cleaned, - feed: feed, - msgLimit: msgLimit, - baseCh: baseCh, + channels: cleaned, + feed: feed, + msgLimit: msgLimit, + baseCh: baseCh, client: &http.Client{ Timeout: 30 * time.Second, }, - baseURL: "https://t.me/s", + baseURL: "https://t.me/s", cache: make(map[string]cachedMessages), cacheTTL: 10 * time.Minute, fetchInterval: 10 * time.Minute, refreshCh: make(chan struct{}, 1), + limits: limits, } } @@ -133,6 +135,12 @@ func (pr *PublicReader) fetchAll(ctx context.Context) { pr.mu.RUnlock() for i, username := range pr.channels { chNum := pr.baseCh + i + ctx := ctx + if pr.limits != nil { + if lim, ok := pr.limits[strings.ToLower(username)]; ok { + ctx = WithContextLimits(ctx, lim.MediaSize, lim.AudioSize) + } + } pr.mu.RLock() cached, ok := pr.cache[username] diff --git a/internal/server/server.go b/internal/server/server.go index 93c8fb2..0edd419 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -8,6 +8,7 @@ import ( "log" "os" "path/filepath" + "strconv" "strings" "time" @@ -35,6 +36,7 @@ type Config struct { // but the wire-format DNS flag is unset for clients. DNSMediaEnabled bool DNSMediaMaxSize int64 // per-file cap for the DNS relay (0 = no cap) + DNSAudioMaxSize int64 // per-file cap for audio/voice files in the DNS relay (0 = fallback) 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 @@ -85,6 +87,7 @@ type Server struct { telegramChannels []string privateInvites []string // resolved invite hashes (post-parse) xAccounts []string + limits map[string]ChannelLimits } // New creates a new Server. @@ -110,6 +113,23 @@ func New(cfg Config) (*Server, error) { cfg.ChannelsFile, cfg.PrivateChannelsFile, cfg.XAccountsFile) } + limits := make(map[string]ChannelLimits) + if chanLims, err := loadLimitsFromFile(cfg.ChannelsFile, false); err == nil { + for k, v := range chanLims { + limits[k] = v + } + } + if privLims, err := loadLimitsFromFile(cfg.PrivateChannelsFile, true); err == nil { + for k, v := range privLims { + limits[k] = v + } + } + if xLims, err := loadLimitsFromFile(cfg.XAccountsFile, false); err == nil { + for k, v := range xLims { + limits[k] = v + } + } + log.Printf("[server] loaded %d Telegram public channels, %d private invites, %d X accounts", len(channels), len(privateInvites), len(xAccounts)) @@ -131,6 +151,7 @@ func New(cfg Config) (*Server, error) { telegramChannels: channels, privateInvites: privateInvites, xAccounts: xAccounts, + limits: limits, }, nil } @@ -247,6 +268,7 @@ func (s *Server) Run(ctx context.Context) error { } mediaCache := NewMediaCache(MediaCacheConfig{ MaxFileBytes: s.cfg.DNSMediaMaxSize, + MaxAudioBytes: s.cfg.DNSAudioMaxSize, TTL: ttl, Compression: compression, Logf: logfMedia, @@ -280,7 +302,7 @@ func (s *Server) Run(ctx context.Context) error { // Handle login-only mode if s.cfg.Telegram.LoginOnly { - reader := NewTelegramReader(s.cfg.Telegram, s.telegramChannels, s.privateInvites, s.feed, 15, 1) + reader := NewTelegramReader(s.cfg.Telegram, s.telegramChannels, s.privateInvites, s.feed, 15, 1, s.limits) return reader.Run(ctx) } @@ -291,7 +313,7 @@ func (s *Server) Run(ctx context.Context) error { msgLimit = 15 } if len(s.telegramChannels) > 0 || len(s.privateInvites) > 0 { - reader := NewTelegramReader(s.cfg.Telegram, s.telegramChannels, s.privateInvites, s.feed, msgLimit, 1) + reader := NewTelegramReader(s.cfg.Telegram, s.telegramChannels, s.privateInvites, s.feed, msgLimit, 1, s.limits) reader.SetFetchInterval(s.cfg.FetchInterval) s.reader = reader channelCtl = reader @@ -311,7 +333,7 @@ func (s *Server) Run(ctx context.Context) error { if msgLimit <= 0 { msgLimit = 15 } - publicReader := NewPublicReader(s.telegramChannels, s.feed, msgLimit, 1) + publicReader := NewPublicReader(s.telegramChannels, s.feed, msgLimit, 1, s.limits) publicReader.SetFetchInterval(s.cfg.FetchInterval) channelCtl = publicReader go func() { @@ -332,7 +354,7 @@ func (s *Server) Run(ctx context.Context) error { msgLimit = 15 } // X channel numbers start after all Telegram channels (public + private). - xReader = NewXPublicReader(s.xAccounts, s.feed, msgLimit, len(s.telegramChannels)+len(s.privateInvites)+1, s.cfg.XRSSInstances) + xReader = NewXPublicReader(s.xAccounts, s.feed, msgLimit, len(s.telegramChannels)+len(s.privateInvites)+1, s.cfg.XRSSInstances, s.limits) xReader.SetFetchInterval(s.cfg.FetchInterval) go func() { log.Println("[x] reader goroutine started") @@ -370,6 +392,83 @@ func (s *Server) Run(ctx context.Context) error { return dnsServer.ListenAndServe(ctx) } +type ChannelLimits struct { + MediaSize int64 // bytes, -1 if not set + AudioSize int64 // bytes, -1 if not set +} + +type contextLimitsKey struct{} + +type Limits struct { + MaxFileBytes int64 + MaxAudioBytes int64 +} + +func WithContextLimits(ctx context.Context, maxFile, maxAudio int64) context.Context { + return context.WithValue(ctx, contextLimitsKey{}, Limits{ + MaxFileBytes: maxFile, + MaxAudioBytes: maxAudio, + }) +} + +func GetContextLimits(ctx context.Context) (int64, int64, bool) { + if ctx == nil { + return 0, 0, false + } + val := ctx.Value(contextLimitsKey{}) + if val == nil { + return 0, 0, false + } + lims := val.(Limits) + return lims.MaxFileBytes, lims.MaxAudioBytes, true +} + +func GetMaxBytesFromContext(ctx context.Context, tag, mimeType string, cache *MediaCache) (int64, bool) { + if cache == nil { + return 0, false + } + maxFile, maxAudio, ok := GetContextLimits(ctx) + if !ok { + return 0, false + } + + dns := maxFile + if dns == -1 { + dns = cache.maxFileBytes + } + + dnsAudio := maxAudio + if dnsAudio == -1 { + dnsAudio = cache.maxAudioBytes + } + + if dnsAudio > 0 && isAudioOrVoice(tag, mimeType) { + dns = dnsAudio + } + + var ghMax int64 + cache.mu.RLock() + gh := cache.gh + cache.mu.RUnlock() + if gh != nil { + ghMax = gh.MaxBytes() + } + + if (dns == 0 && cache.dnsEnabled) || (gh != nil && ghMax == 0) { + return 0, true + } + if !cache.dnsEnabled { + return ghMax, true + } + if gh == nil { + return dns, true + } + if ghMax > dns { + return ghMax, true + } + return dns, true +} + func loadUsernames(path string) ([]string, error) { f, err := os.Open(path) if err != nil { @@ -391,12 +490,76 @@ func loadUsernames(path string) ([]string, error) { if line == "" || strings.HasPrefix(line, "#") { continue } - name := strings.TrimPrefix(line, "@") - users = append(users, name) + parts := strings.Fields(line) + if len(parts) > 0 { + name := strings.TrimPrefix(parts[0], "@") + users = append(users, name) + } } return users, scanner.Err() } +func loadLimitsFromFile(path string, isPrivate bool) (map[string]ChannelLimits, error) { + f, err := os.Open(path) + if err != nil { + if os.IsNotExist(err) { + return make(map[string]ChannelLimits), nil + } + return nil, err + } + defer f.Close() + + limits := make(map[string]ChannelLimits) + scanner := bufio.NewScanner(f) + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if line == "" || strings.HasPrefix(line, "#") { + continue + } + parts := strings.Fields(line) + if len(parts) == 0 { + continue + } + + var name string + if isPrivate { + hash, err := ParseInviteHash(parts[0]) + if err != nil { + continue + } + name = hash + } else { + name = strings.TrimPrefix(parts[0], "@") + name = strings.TrimPrefix(name, "x/") + name = strings.ToLower(name) + } + + var mediaSize, audioSize int64 = -1, -1 + for i := 1; i < len(parts); i++ { + if parts[i] == "--dns-media-max-size" && i+1 < len(parts) { + val, err := strconv.ParseInt(parts[i+1], 10, 64) + if err == nil { + mediaSize = val * 1024 + } + i++ + } else if parts[i] == "--dns-audio-max-size" && i+1 < len(parts) { + val, err := strconv.ParseInt(parts[i+1], 10, 64) + if err == nil { + audioSize = val * 1024 + } + i++ + } + } + if mediaSize != -1 || audioSize != -1 { + limits[name] = ChannelLimits{ + MediaSize: mediaSize, + AudioSize: audioSize, + } + } + } + return limits, scanner.Err() +} + func prefixXAccounts(accounts []string) []string { out := make([]string, len(accounts)) for i, a := range accounts { diff --git a/internal/server/telegram.go b/internal/server/telegram.go index 0a8a70a..05d6f05 100644 --- a/internal/server/telegram.go +++ b/internal/server/telegram.go @@ -83,6 +83,8 @@ type TelegramReader struct { api *tg.Client refreshCh chan struct{} // signals Run() to re-fetch immediately + + limits map[string]ChannelLimits } // SetFetchInterval overrides the default 10m fetch cadence. @@ -117,7 +119,7 @@ type cachedMessages struct { // and private-channel invite hashes. privateInviteHashes is the list // of base64url-ish invite codes (see ParseInviteHash) — empty when no // private channels are configured. -func NewTelegramReader(cfg TelegramConfig, channelUsernames []string, privateInviteHashes []string, feed *Feed, msgLimit int, baseCh int) *TelegramReader { +func NewTelegramReader(cfg TelegramConfig, channelUsernames []string, privateInviteHashes []string, feed *Feed, msgLimit int, baseCh int, limits map[string]ChannelLimits) *TelegramReader { cleaned := make([]string, len(channelUsernames)) for i, u := range channelUsernames { cleaned[i] = strings.TrimPrefix(strings.TrimSpace(u), "@") @@ -139,6 +141,7 @@ func NewTelegramReader(cfg TelegramConfig, channelUsernames []string, privateInv cacheTTL: 10 * time.Minute, fetchInterval: 10 * time.Minute, refreshCh: make(chan struct{}, 1), + limits: limits, } } @@ -269,6 +272,12 @@ func (tr *TelegramReader) fetchAll(ctx context.Context, api *tg.Client) { tr.mu.RUnlock() for i, username := range tr.channels { chNum := tr.baseCh + i + ctx := ctx + if tr.limits != nil { + if lim, ok := tr.limits[strings.ToLower(username)]; ok { + ctx = WithContextLimits(ctx, lim.MediaSize, lim.AudioSize) + } + } tr.mu.RLock() cached, ok := tr.cache[username] @@ -321,6 +330,12 @@ func (tr *TelegramReader) fetchAll(ctx context.Context, api *tg.Client) { // Cache key is the short hash-derived channel ID. for i, hash := range tr.privates.ordered { chNum := tr.baseCh + len(tr.channels) + i + ctx := ctx + if tr.limits != nil { + if lim, ok := tr.limits[hash]; ok { + ctx = WithContextLimits(ctx, lim.MediaSize, lim.AudioSize) + } + } rp, ok := tr.privates.get(hash) if !ok { // Retry resolve if startup failed. diff --git a/internal/server/xpublic.go b/internal/server/xpublic.go index bce7b6c..c92a478 100644 --- a/internal/server/xpublic.go +++ b/internal/server/xpublic.go @@ -38,6 +38,7 @@ type XPublicReader struct { fetchInterval time.Duration refreshCh chan struct{} + limits map[string]ChannelLimits } // SetFetchInterval overrides the default 10m fetch cadence. @@ -58,7 +59,7 @@ var xHandleRe = regexp.MustCompile(`@[A-Za-z0-9_]{1,15}`) var xRepostLeadRe = regexp.MustCompile(`(?i)^RT\s+(@[A-Za-z0-9_]{1,15})\s*:\s*`) var xRepostByRe = regexp.MustCompile(`(?i)^RT\s+by\s+.+\((@[A-Za-z0-9_]{1,15})\)\s*:\s*`) -func NewXPublicReader(accounts []string, feed *Feed, msgLimit int, baseCh int, instancesCSV string) *XPublicReader { +func NewXPublicReader(accounts []string, feed *Feed, msgLimit int, baseCh int, instancesCSV string, limits map[string]ChannelLimits) *XPublicReader { cleaned := make([]string, 0, len(accounts)) for _, a := range accounts { a = strings.TrimSpace(strings.TrimPrefix(a, "@")) @@ -91,6 +92,7 @@ func NewXPublicReader(accounts []string, feed *Feed, msgLimit int, baseCh int, i cacheTTL: 10 * time.Minute, fetchInterval: 10 * time.Minute, refreshCh: make(chan struct{}, 1), + limits: limits, } } @@ -189,6 +191,12 @@ func (xr *XPublicReader) fetchAll(ctx context.Context) { for i, account := range xr.accounts { chNum := baseCh + i + ctx := ctx + if xr.limits != nil { + if lim, ok := xr.limits[strings.ToLower(account)]; ok { + ctx = WithContextLimits(ctx, lim.MediaSize, lim.AudioSize) + } + } xr.mu.RLock() cached, ok := xr.cache[account]