icyws

a tiny ices2 input module+websocket
Download | Log | Files | Refs | README | LICENSE

main.go (8888B)

      1 // icyws - an ices2 input module which broadcasts metadata over websockets
      2 // Copyright (c) 2025 Andrew Kloet
      3 //
      4 // This program is free software: you can redistribute it and/or modify
      5 // it under the terms of the GNU Affero General Public License as published by
      6 // the Free Software Foundation, either version 3 of the License, or
      7 // (at your option) any later version.
      8 //
      9 // This program is distributed in the hope that it will be useful,
     10 // but WITHOUT ANY WARRANTY; without even the implied warranty of
     11 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
     12 // GNU Affero General Public License for more details.
     13 //
     14 // You should have received a copy of the GNU Affero General Public License
     15 // along with this program.  If not, see <https://www.gnu.org/licenses/>.
     16 
     17 package main
     18 
     19 import (
     20 	"bufio"
     21 	"bytes"
     22 	"context"
     23 	"encoding/json"
     24 	"errors"
     25 	"flag"
     26 	"fmt"
     27 	"io"
     28 	"log"
     29 	"math/rand"
     30 	"net/http"
     31 	"os"
     32 	"os/signal"
     33 	"path/filepath"
     34 	"strings"
     35 	"sync"
     36 	"syscall"
     37 	"time"
     38 
     39 	"github.com/dhowden/tag"
     40 	"github.com/gorilla/websocket"
     41 )
     42 
     43 const (
     44 	defaultBindAddress = ":8765"
     45 	announceEndpoint   = "http://127.0.0.1:8765/announce"
     46 )
     47 
     48 var rng = rand.New(rand.NewSource(time.Now().UnixNano()))
     49 
     50 type trackInfo struct {
     51 	Path   string
     52 	Artist string
     53 	Title  string
     54 }
     55 
     56 func main() {
     57 	log.SetFlags(0)
     58 
     59 	if len(os.Args) > 1 && os.Args[1] == "serve" {
     60 		serveCmd(os.Args[2:])
     61 		return
     62 	}
     63 
     64 	pickCmd(os.Args[1:])
     65 }
     66 
     67 func pickCmd(args []string) {
     68 	fs := flag.NewFlagSet("pick", flag.ExitOnError)
     69 	tracksPath := fs.String("tracks", os.Getenv("ICYWS_TRACKLIST"), "path to track list (newline separated)")
     70 	fs.Parse(args)
     71 
     72 	if *tracksPath == "" {
     73 		log.Fatal("missing track list: set ICYWS_TRACKLIST or pass --tracks")
     74 	}
     75 
     76 	info, err := pickRandomTrack(*tracksPath)
     77 	if err != nil {
     78 		log.Fatalf("selecting track: %v", err)
     79 	}
     80 
     81 	fmt.Println(info.Path)
     82 
     83 	nowPlaying := formatNowPlaying(info.Artist, info.Title)
     84 	if err := announceNowPlaying(nowPlaying); err != nil {
     85 		fmt.Fprintf(os.Stderr, "warn: %v\n", err)
     86 	}
     87 }
     88 
     89 func pickRandomTrack(listPath string) (trackInfo, error) {
     90 	file, err := os.Open(listPath)
     91 	if err != nil {
     92 		return trackInfo{}, fmt.Errorf("open track list: %w", err)
     93 	}
     94 	defer file.Close()
     95 
     96 	scanner := bufio.NewScanner(file)
     97 	const maxLine = 1024 * 1024
     98 	scanner.Buffer(make([]byte, 0, 64*1024), maxLine)
     99 
    100 	var tracks []string
    101 	for scanner.Scan() {
    102 		line := strings.TrimSpace(scanner.Text())
    103 		if line == "" || strings.HasPrefix(line, "#") {
    104 			continue
    105 		}
    106 		tracks = append(tracks, line)
    107 	}
    108 	if err := scanner.Err(); err != nil {
    109 		return trackInfo{}, fmt.Errorf("read track list: %w", err)
    110 	}
    111 	if len(tracks) == 0 {
    112 		return trackInfo{}, errors.New("track list is empty")
    113 	}
    114 
    115 	path := tracks[rng.Intn(len(tracks))]
    116 	artist, title := extractMetadata(path)
    117 
    118 	return trackInfo{
    119 		Path:   path,
    120 		Artist: artist,
    121 		Title:  title,
    122 	}, nil
    123 }
    124 
    125 func extractMetadata(path string) (string, string) {
    126 	file, err := os.Open(path)
    127 	if err != nil {
    128 		return "Unknown Artist", fallbackTitle(path)
    129 	}
    130 	defer file.Close()
    131 
    132 	meta, err := tag.ReadFrom(file)
    133 	if err != nil {
    134 		return "Unknown Artist", fallbackTitle(path)
    135 	}
    136 
    137 	artist := meta.Artist()
    138 	if artist == "" {
    139 		artist = meta.AlbumArtist()
    140 	}
    141 	if artist == "" {
    142 		artist = "Unknown Artist"
    143 	}
    144 
    145 	title := meta.Title()
    146 	if title == "" {
    147 		title = fallbackTitle(path)
    148 	}
    149 
    150 	return artist, title
    151 }
    152 
    153 func fallbackTitle(path string) string {
    154 	base := filepath.Base(path)
    155 	ext := filepath.Ext(base)
    156 	return strings.TrimSuffix(base, ext)
    157 }
    158 
    159 func formatNowPlaying(artist, title string) string {
    160 	return fmt.Sprintf("%s - %s", artist, title)
    161 }
    162 
    163 func announceNowPlaying(nowPlaying string) error {
    164 	ctx, cancel := context.WithTimeout(context.Background(), 750*time.Millisecond)
    165 	defer cancel()
    166 
    167 	payload := map[string]string{"now_playing": nowPlaying}
    168 	body, _ := json.Marshal(payload)
    169 
    170 	req, err := http.NewRequestWithContext(ctx, http.MethodPost, announceEndpoint, bytes.NewReader(body))
    171 	if err != nil {
    172 		return fmt.Errorf("prepare announce request: %w", err)
    173 	}
    174 	req.Header.Set("Content-Type", "application/json")
    175 
    176 	resp, err := http.DefaultClient.Do(req)
    177 	if err != nil {
    178 		return fmt.Errorf("broadcast failed (is icyws serve running?): %w", err)
    179 	}
    180 	defer resp.Body.Close()
    181 
    182 	if resp.StatusCode >= 400 {
    183 		b, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
    184 		return fmt.Errorf("broadcast rejected: %s %s", resp.Status, strings.TrimSpace(string(b)))
    185 	}
    186 
    187 	return nil
    188 }
    189 
    190 func serveCmd(args []string) {
    191 	fs := flag.NewFlagSet("serve", flag.ExitOnError)
    192 	bindAddr := fs.String("bind", defaultBindAddress, "address:port to listen on")
    193 	allowOrigin := fs.String("allow-origin", "", "restrict websocket Origin (optional)")
    194 	fs.Parse(args)
    195 
    196 	server := newNowPlayingServer(*allowOrigin)
    197 
    198 	mux := http.NewServeMux()
    199 	mux.HandleFunc("/radio/ws", server.handleWS)
    200 	mux.HandleFunc("/announce", server.handleAnnounce)
    201 	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
    202 		w.WriteHeader(http.StatusOK)
    203 		w.Write([]byte("ok"))
    204 	})
    205 
    206 	httpServer := &http.Server{
    207 		Addr:              *bindAddr,
    208 		Handler:           logRequests(mux),
    209 		ReadHeaderTimeout: 5 * time.Second,
    210 	}
    211 
    212 	shutdownCtx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
    213 	defer stop()
    214 
    215 	go func() {
    216 		<-shutdownCtx.Done()
    217 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    218 		defer cancel()
    219 		httpServer.Shutdown(ctx)
    220 	}()
    221 
    222 	log.Printf("icyws serve listening on %s", *bindAddr)
    223 	if err := httpServer.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
    224 		log.Fatalf("server error: %v", err)
    225 	}
    226 	log.Println("icyws serve stopped")
    227 }
    228 
    229 type nowPlayingServer struct {
    230 	mu          sync.RWMutex
    231 	clients     map[*websocket.Conn]struct{}
    232 	lastMessage string
    233 	upgrader    websocket.Upgrader
    234 }
    235 
    236 func newNowPlayingServer(allowedOrigin string) *nowPlayingServer {
    237 	upgrader := websocket.Upgrader{
    238 		CheckOrigin: func(r *http.Request) bool {
    239 			if allowedOrigin == "" {
    240 				return true
    241 			}
    242 			return r.Header.Get("Origin") == allowedOrigin
    243 		},
    244 	}
    245 	return &nowPlayingServer{
    246 		clients:  make(map[*websocket.Conn]struct{}),
    247 		upgrader: upgrader,
    248 	}
    249 }
    250 
    251 func (s *nowPlayingServer) handleWS(w http.ResponseWriter, r *http.Request) {
    252 	conn, err := s.upgrader.Upgrade(w, r, nil)
    253 	if err != nil {
    254 		log.Printf("websocket upgrade: %v", err)
    255 		return
    256 	}
    257 
    258 	s.addClient(conn)
    259 
    260 	if last := s.getLast(); last != "" {
    261 		_ = conn.WriteMessage(websocket.TextMessage, []byte(last))
    262 	}
    263 
    264 	go s.readUntilClose(conn)
    265 }
    266 
    267 func (s *nowPlayingServer) handleAnnounce(w http.ResponseWriter, r *http.Request) {
    268 	if r.Method != http.MethodPost {
    269 		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
    270 		return
    271 	}
    272 	defer r.Body.Close()
    273 
    274 	body, err := io.ReadAll(io.LimitReader(r.Body, 2048))
    275 	if err != nil {
    276 		http.Error(w, "read body", http.StatusBadRequest)
    277 		return
    278 	}
    279 
    280 	var payload struct {
    281 		NowPlaying string `json:"now_playing"`
    282 	}
    283 
    284 	if len(bytes.TrimSpace(body)) == 0 {
    285 		http.Error(w, "empty payload", http.StatusBadRequest)
    286 		return
    287 	}
    288 
    289 	if err := json.Unmarshal(body, &payload); err != nil {
    290 		payload.NowPlaying = strings.TrimSpace(string(body))
    291 	}
    292 
    293 	payload.NowPlaying = strings.TrimSpace(payload.NowPlaying)
    294 	if payload.NowPlaying == "" {
    295 		http.Error(w, "missing now_playing", http.StatusBadRequest)
    296 		return
    297 	}
    298 
    299 	s.setLast(payload.NowPlaying)
    300 	s.broadcast(payload.NowPlaying)
    301 
    302 	w.WriteHeader(http.StatusAccepted)
    303 	w.Write([]byte("ok"))
    304 }
    305 
    306 func (s *nowPlayingServer) addClient(conn *websocket.Conn) {
    307 	s.mu.Lock()
    308 	s.clients[conn] = struct{}{}
    309 	s.mu.Unlock()
    310 }
    311 
    312 func (s *nowPlayingServer) removeClient(conn *websocket.Conn) {
    313 	s.mu.Lock()
    314 	delete(s.clients, conn)
    315 	s.mu.Unlock()
    316 	conn.Close()
    317 }
    318 
    319 func (s *nowPlayingServer) broadcast(message string) {
    320 	s.mu.RLock()
    321 	conns := make([]*websocket.Conn, 0, len(s.clients))
    322 	for c := range s.clients {
    323 		conns = append(conns, c)
    324 	}
    325 	s.mu.RUnlock()
    326 
    327 	for _, conn := range conns {
    328 		if err := conn.WriteMessage(websocket.TextMessage, []byte(message)); err != nil {
    329 			s.removeClient(conn)
    330 		}
    331 	}
    332 }
    333 
    334 func (s *nowPlayingServer) getLast() string {
    335 	s.mu.RLock()
    336 	defer s.mu.RUnlock()
    337 	return s.lastMessage
    338 }
    339 
    340 func (s *nowPlayingServer) setLast(message string) {
    341 	s.mu.Lock()
    342 	s.lastMessage = message
    343 	s.mu.Unlock()
    344 }
    345 
    346 func (s *nowPlayingServer) readUntilClose(conn *websocket.Conn) {
    347 	defer s.removeClient(conn)
    348 	conn.SetReadLimit(1024)
    349 	conn.SetReadDeadline(time.Now().Add(3600 * time.Second))
    350 	conn.SetPongHandler(func(string) error {
    351 		conn.SetReadDeadline(time.Now().Add(3600 * time.Second))
    352 		return nil
    353 	})
    354 
    355 	for {
    356 		if _, _, err := conn.ReadMessage(); err != nil {
    357 			break
    358 		}
    359 	}
    360 }
    361 
    362 func logRequests(next http.Handler) http.Handler {
    363 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
    364 		start := time.Now()
    365 		next.ServeHTTP(w, r)
    366 		log.Printf("%s %s %s", r.Method, r.URL.Path, time.Since(start).Truncate(time.Millisecond))
    367 	})
    368 }