mcg/internal/client/client.go

package client

import (
	"fmt"
	"io"
	"log"
	"net/textproto"
	"strconv"
	"strings"
	"sync"
	"time"

	"suruatoel.xyz/mcg/internal/common"
)

type QueueItem struct {
	Command func(*Client) error
}

type Client struct {
	Host            string
	con             *textproto.Conn
	runGroup        sync.WaitGroup
	queue           chan QueueItem
	idling          bool
	version         string
	state           string
	playlist        []*common.PlaylistAlbum
	albums          map[string]*common.Album
	errorCB         ErrorCB
	connectionCB    ConnectionCB
	playlistCB      PlaylistCB
	albumsCB        AlbumsCB
	statusCB        StatusCB
	statisticsCB    StatisticsCB
	outputDevicesCB OutputDevicesCB
	albumartCB      AlbumartCB
	albumartChan    chan []byte
}

type ErrorCB func(string)

type ConnectionCB func(bool)

type PlaylistCB func([]*common.PlaylistAlbum)

type AlbumsCB func(map[string]*common.Album)

type StatusCB func(string, *common.PlaylistAlbum, uint64, uint64, int64, string, string, string, string)

type StatisticsCB func(string, string, string, string, string, string)

type OutputDevicesCB func(devices []*common.OutputDevice)

type AlbumartCB func(*common.Album, []byte)

var (
	clientInstance *Client
	clientInit     sync.Once
)

func Instance() *Client {
	clientInit.Do(func() {
		clientInstance = &Client{}
	})

	return clientInstance
}

func (c *Client) IsConnected() bool {
	return (c.con != nil)
}

func (c *Client) Connect(host string, port int, password string) {
	c.Host = host
	c.queue = make(chan QueueItem, 300)
	c.albumartChan = make(chan []byte)
	// Add connect action
	c.addAction(
		func(client *Client) error {
			return client.connect(host, port, password)
		},
	)

	go c.run()
}

func (c *Client) connect(host string, port int, password string) error {
	// Notify listeners
	defer func() {
		if c.connectionCB != nil {
			c.connectionCB(c.con != nil)
		}
	}()

	// Create socket
	con, err := textproto.Dial("tcp", fmt.Sprintf("%s:%d", host, port))
	c.con = con
	if err != nil {
		return err
	}

	// Greeting
	greeting, err := c.readLine()
	if err != nil {
		return err
	}
	if greeting[0:6] != "OK MPD" {
		return textproto.ProtocolError(fmt.Sprintf("Unexpected greeting: %s", greeting))
	}

	// Parse and store server version
	c.version = greeting[7:]

	// Add post-connect actions
	c.addPostConnectActions(password)

	return nil
}

func (c *Client) addPostConnectActions(password string) {
	c.addAction(
		func(client *Client) error {
			return client.password(password)
		},
		func(client *Client) error {
			return client.loadAlbums()
		},
		func(client *Client) error {
			return client.loadPlaylist()
		},
		func(client *Client) error {
			return client.getStatus()
		},
		func(client *Client) error {
			return client.getStatistics()
		},
		func(client *Client) error {
			return client.getOutputDevices()
		},
	)
}

func (c *Client) password(password string) error {
	if password != "" {
		reader := &emptyReader{}
		return c.cmd(reader, "password", password)
	}

	return nil
}

func (c *Client) Disconnect() {
	c.addDisconnectActions()

	close(c.queue)
	close(c.albumartChan)
	c.runGroup.Wait()
}

func (c *Client) addDisconnectActions() {
	c.addAction(
		func(client *Client) error {
			return client.disconnect()
		},
	)
}

func (c *Client) disconnect() error {
	// Stop current idle process
	c.noIdle()

	// Close connection
	err := c.con.Close()
	if err != nil {
		return err
	}
	c.con = nil

	// Notify callback
	if c.connectionCB != nil {
		c.connectionCB(false)
	}

	return nil
}

func (c *Client) run() {
	c.runGroup.Add(1)
	defer func() {
		c.runGroup.Done()
	}()

	for action := range c.queue {
		err := action.Command(c)
		if err != nil && c.errorCB != nil {
			c.errorCB(fmt.Sprintf("Error running action: %s\n", err))
		}
	}
}

func (c *Client) Version() string {
	return c.version
}

func (c *Client) ReceiveError(callback ErrorCB) {
	c.errorCB = callback
}

func (c *Client) RegisterConnection(callback ConnectionCB) {
	c.connectionCB = callback
}

func (c *Client) ReceivePlaylist(callback PlaylistCB) {
	c.playlistCB = callback
}

func (c *Client) ReceiveAlbums(callback AlbumsCB) {
	c.albumsCB = callback
}

func (c *Client) ReceiveStatus(callback StatusCB) {
	c.statusCB = callback
}

func (c *Client) ReceiveStatistics(callback StatisticsCB) {
	c.statisticsCB = callback
}

func (c *Client) ReceiveOutputDevices(callback OutputDevicesCB) {
	c.outputDevicesCB = callback
}

func (c *Client) ReceiveAlbumart(callback AlbumartCB) {
	c.albumartCB = callback
}

func (c *Client) loadPlaylist() error {
	c.playlist = []*common.PlaylistAlbum{}

	playlistReader := &listMapReader{Delimiter: "file"}
	playlistErr := c.cmd(playlistReader, "playlistinfo")
	if playlistErr != nil {
		return playlistErr
	}

	for _, songData := range playlistReader.DictList() {
		song := c.extractPlaylistSong(songData)
		album := c.extractPlaylistAlbum(songData)
		if len(c.playlist) == 0 || c.playlist[len(c.playlist)-1].ID() != album.ID() {
			c.playlist = append(c.playlist, album)
		} else {
			album = c.playlist[len(c.playlist)-1]
		}
		if song != nil {
			album.AddSong(song)
		}
	}

	if c.playlistCB != nil {
		c.playlistCB(c.playlist)
	}

	return nil
}

func (c *Client) loadAlbums() error {
	// self._callback(Client.SIGNAL_INIT_ALBUMS)
	c.albums = make(map[string]*common.Album)

	albumReader := &listMapReader{Delimiter: "Album"}
	albumErr := c.cmd(albumReader, "list album")
	if albumErr != nil {
		return albumErr
	}
	for _, entry := range albumReader.DictList() {
		// self._callback(Client.SIGNAL_PULSE_ALBUMS)

		// Album
		album := c.extractAlbum(entry)
		if album != nil {
			// Songs
			songReader := &listMapReader{Delimiter: "file"}
			songErr := c.cmd(songReader, "find album", album.Title())
			if songErr != nil {
				return songErr
			}
			for _, songData := range songReader.DictList() {
				song := c.extractSong(songData)
				if song != nil {
					album.AddSong(song)
				}
			}

			c.albums[album.ID()] = album
		}
	}

	if c.albumsCB != nil {
		c.albumsCB(c.albums)
	}

	return nil
}

func (c *Client) getStatus() error {
	statusReader := &mapReader{}
	statusErr := c.cmd(statusReader, "status")
	if statusErr != nil {
		return statusErr
	}

	status := statusReader.Dict()
	// State
	c.state = status["state"]
	// Time
	var parseErr error
	time := uint64(0)
	if timeStr, timeSet := status["time"]; timeSet {
		time, parseErr = strconv.ParseUint(strings.Split(timeStr, ":")[0], 10, 64)
		if parseErr != nil {
			return newValueParseError("time", timeStr)
		}
	}
	// Volume
	volume := int64(-1)
	if volumeStr, volumeSet := status["volume"]; volumeSet {
		volume, parseErr = strconv.ParseInt(volumeStr, 10, 64)
		if parseErr != nil {
			return newValueParseError("volume", volumeStr)
		}
	}
	// Error
	errorMessage := status["error"]

	// Album information
	file := ""
	var album *common.PlaylistAlbum
	pos := uint64(0)
	songReader := &mapReader{}
	songErr := c.cmd(songReader, "currentsong")
	if songErr != nil {
		return songErr
	}
	songData := songReader.Dict()
	if len(songData) > 0 {
		// File
		file = songData["file"]
		// Song
		song := c.extractPlaylistSong(songData)
		if song != nil {
			// Album
			album = c.extractPlaylistAlbum(songData)
			pos = song.Pos()
			for _, palbum := range c.playlist {
				if palbum.ID() == album.ID() && uint64(len(palbum.Songs())) >= pos {
					album = palbum
					break
				}
				pos -= uint64(len(palbum.Songs()))
			}
		}
	}
	// Audio
	audio := status["audio"]
	// Bitrate
	bitrate := status["bitrate"]

	if c.statusCB != nil {
		c.statusCB(c.state, album, pos, time, volume, file, audio, bitrate, errorMessage)
	}

	return nil
}

func (c *Client) getStatistics() error {
	statisticsReader := &mapReader{}
	statisticsErr := c.cmd(statisticsReader, "stats")
	if statisticsErr != nil {
		return statisticsErr
	}

	stats := statisticsReader.Dict()
	artists := stats["artists"]
	albums := stats["albums"]
	songs := stats["songs"]
	dbPlaytime := stats["db_playtime"]
	playtime := stats["playtime"]
	uptime := stats["uptime"]

	if c.statisticsCB != nil {
		c.statisticsCB(artists, albums, songs, dbPlaytime, playtime, uptime)
	}

	return nil
}

func (c *Client) getOutputDevices() error {
	devicesReader := &listMapReader{Delimiter: "outputid"}
	devicesErr := c.cmd(devicesReader, "outputs")
	if devicesErr != nil {
		return devicesErr
	}

	devices := []*common.OutputDevice{}
	for _, deviceMap := range devicesReader.DictList() {
		device := common.NewOutputDevice(
			deviceMap["outputid"],
			deviceMap["outputname"],
		)

		enabledString := deviceMap["outputenabled"]
		if enabledString != "" {
			enabled, enabledErr := strconv.ParseBool(enabledString)
			if enabledErr == nil {
				device.Enabled = enabled
			} else {
				log.Println("failed to parse status of output device", device.ID, device.Name)
			}
		}

		devices = append(devices, device)
	}

	if c.outputDevicesCB != nil {
		c.outputDevicesCB(devices)
	}

	return nil
}

func (c *Client) SetOutputDeviceState(device *common.OutputDevice, enable bool) {
	c.addAction(func(client *Client) error {
		return client.setOutputDeviceState(device, enable)
	})
}

func (c *Client) setOutputDeviceState(device *common.OutputDevice, enable bool) error {
	reader := &emptyReader{}
	if enable {
		return c.cmd(reader, "enableoutput", device.ID)
	}
	return c.cmd(reader, "disableoutput", device.ID)
}

func (c *Client) PlayPause() {
	c.addAction(func(client *Client) error {
		return client.playPause()
	})
}

func (c *Client) playPause() error {
	reader := &emptyReader{}
	switch c.state {
	case "play":
		return c.cmd(reader, "pause")
	default:
		return c.cmd(reader, "play")
	}
}

func (c *Client) PlayAlbum(album *common.Album) {
	c.addAction(func(client *Client) error {
		return client.playAlbum(album)
	})
}

func (c *Client) playAlbum(album *common.Album) error {
	songIDs, err := c.queueAlbum(album)
	if err != nil {
		return err
	}

	if len(songIDs) > 0 {
		reader := &emptyReader{}
		return c.cmd(reader, "playid", songIDs[0])
	}

	return nil
}

func (c *Client) QueueAlbum(album *common.Album) {
	c.addAction(func(client *Client) error {
		_, err := client.queueAlbum(album)
		return err
	})
}

func (c *Client) queueAlbum(album *common.Album) ([]string, error) {
	songIDs := []string{}
	for _, song := range album.Songs() {
		reader := &mapReader{}
		addErr := c.cmd(reader, "addid", song.File())
		if addErr != nil {
			return songIDs, addErr
		}

		addData := reader.Dict()
		songID, found := addData["Id"]
		if found {
			songIDs = append(songIDs, songID)
		}
	}

	return songIDs, nil
}

func (c *Client) QueueAlbums(albums []*common.Album) {
	c.addAction(func(client *Client) error {
		return client.queueAlbums(albums)
	})
}

func (c *Client) queueAlbums(albums []*common.Album) error {
	for _, album := range albums {
		_, queueErr := c.queueAlbum(album)
		if queueErr != nil {
			return queueErr
		}
	}

	return nil
}

func (c *Client) PlayQueuedAlbum(album *common.PlaylistAlbum) {
	c.addAction(func(client *Client) error {
		return client.playQueuedAlbum(album)
	})
}

func (c *Client) playQueuedAlbum(album *common.PlaylistAlbum) error {
	if len(album.Songs()) <= 0 {
		return nil
	}

	reader := &emptyReader{}
	return c.cmd(reader, "playid", album.Songs()[0].ID())
}

func (c *Client) UnqueueAlbum(album *common.PlaylistAlbum) {
	c.addAction(func(client *Client) error {
		return client.unqueueAlbum(album)
	})
}

func (c *Client) unqueueAlbum(album *common.PlaylistAlbum) error {
	commands := []listCommand{}
	for _, song := range album.Songs() {
		command := newListCommand("deleteid", song.ID())
		commands = append(commands, command)
	}

	return c.cmdList(commands)
}

func (c *Client) UnqueueAlbums(albums []*common.PlaylistAlbum) {
	c.addAction(func(client *Client) error {
		return client.unqueueAlbums(albums)
	})
}

func (c *Client) unqueueAlbums(albums []*common.PlaylistAlbum) error {
	commands := []listCommand{}
	for _, album := range albums {
		for _, song := range album.Songs() {
			command := newListCommand("deleteid", song.ID())
			commands = append(commands, command)
		}
	}

	return c.cmdList(commands)
}

func (c *Client) Seek(pos int, time uint64) {
	c.addAction(func(client *Client) error {
		return client.seek(pos, time)
	})
}

func (c *Client) seek(pos int, time uint64) error {
	reader := &emptyReader{}
	return c.cmd(reader, "seek", strconv.Itoa(pos), strconv.FormatUint(time, 10))
}

func (c *Client) SetVolume(volume int) {
	c.addAction(func(client *Client) error {
		return client.setVolume(volume)
	})
}

func (c *Client) setVolume(volume int) error {
	reader := &emptyReader{}
	return c.cmd(reader, "setvol", strconv.Itoa(volume))
}

func (c *Client) LoadAlbumart(album *common.Album) {
	c.addAction(func(client *Client) error {
		return client.loadAlbumart(album)
	})
}

func (c *Client) LoadAlbumartNow(album *common.Album) []byte {
	c.addAction(func(client *Client) error {
		return client.loadAlbumartNow(album)
	})

	return <-c.albumartChan
}

func (c *Client) loadAlbumart(album *common.Album) error {
	reader := &binaryReader{}
	reader.size = 1
	reader.offset = 0

	if c.albumartCB != nil {
		// We need to wrap this in an anonymous function to make sure the call to reader.Binary() is deferred as well
		defer func() {
			c.albumartCB(album, reader.Binary())
		}()
	}

	if len(album.Songs()) > 0 {
		return c.cmd(reader, "albumart", album.Songs()[0].File(), "0")
	}

	return nil
}

func (c *Client) loadAlbumartNow(album *common.Album) error {
	reader := &binaryReader{}
	reader.size = 1
	reader.offset = 0

	defer func() {
		c.albumartChan <- reader.Binary()
	}()

	if len(album.Songs()) > 0 {
		return c.cmd(reader, "albumart", album.Songs()[0].File(), "0")
	}

	return nil
}

func (c *Client) ClearPlaylist() {
	c.addAction(func(client *Client) error {
		return client.clearPlaylist()
	})
}

func (c *Client) clearPlaylist() error {
	reader := &emptyReader{}
	return c.cmd(reader, "clear")
}

func (c *Client) idle() error {
	reader := &listReader{
		key: "changed",
	}

	c.idling = true
	err := c.cmd(reader, "idle")
	c.idling = false
	if err != nil {
		return err
	}

	for _, subsystem := range reader.List() {
		switch subsystem {
		case "player":
			c.addAction(
				func(client *Client) error {
					return client.loadPlaylist()
				},
				func(client *Client) error {
					return client.getStatus()
				},
			)
		case "mixer":
			fallthrough
		case "output":
			c.addAction(func(client *Client) error {
				return client.getStatus()
			})
		case "playlist":
			c.addAction(func(client *Client) error {
				return client.loadPlaylist()
			})
		case "database":
			fallthrough
		case "update":
			c.addAction(
				func(client *Client) error {
					return client.loadAlbums()
				},
				func(client *Client) error {
					return client.loadPlaylist()
				},
				func(client *Client) error {
					return client.getStatus()
				},
			)
		}
	}

	return nil
}

func (c *Client) noIdle() error {
	id, err := c.write("noidle")
	if err != nil {
		return err
	}

	c.con.StartResponse(id)
	defer c.con.EndResponse(id)

	return nil
}

func (c *Client) addAction(commands ...func(*Client) error) {
	for _, command := range commands {
		c.queue <- QueueItem{
			command,
		}
	}
	c.queue <- QueueItem{
		func(client *Client) error {
			if client.con != nil && len(client.queue) == 0 {
				return client.idle()
			}
			return nil
		},
	}
	if c.idling {
		c.noIdle()
	}
}

func (c *Client) cmd(reader reader, action string, args ...string) error {
	readingComplete := false
	for !readingComplete {
		command := c.constructCommand(action, args...)
		id, err := c.write(command)
		if err != nil {
			return textproto.ProtocolError(fmt.Sprintf("failed to write command \"%s\": %s", action, err))
		}

		complete, newArgs, err := c.readData(id, reader, args)
		if err != nil {
			return textproto.ProtocolError(fmt.Sprintf("failed to read response for command \"%s\": %s", action, err))
		}
		readingComplete = complete
		args = newArgs
	}

	return nil
}

func (c *Client) cmdList(commands []listCommand) error {
	beginReader := &noopReader{}
	beginErr := c.cmd(beginReader, "command_list_begin")
	if beginErr != nil {
		return beginErr
	}

	commandReader := &noopReader{}
	for _, command := range commands {
		commandErr := c.cmd(commandReader, command.Action, command.Args...)
		if commandErr != nil {
			return commandErr
		}
	}

	endReader := &emptyReader{}
	return c.cmd(endReader, "command_list_end")
}

func (c *Client) constructCommand(command string, args ...string) string {
	if len(args) == 0 {
		return command
	}

	cleanArgs := c.escapeCommandArguments(args...)
	joinedArgs := strings.Join(cleanArgs, "\" \"")

	return fmt.Sprintf("%s \"%s\"", command, joinedArgs)
}

func (c *Client) escapeCommandArguments(args ...string) []string {
	for i, arg := range args {
		args[i] = strings.ReplaceAll(arg, "\"", "\\\"")
	}

	return args
}

func (c *Client) write(command string) (uint, error) {
	id := c.con.Next()
	c.con.StartRequest(id)
	defer c.con.EndRequest(id)

	fmt.Fprintf(c.con.W, "%s", command)
	c.con.W.WriteByte('\n')
	err := c.con.W.Flush()
	if err != nil {
		return id, err
	}

	return id, nil
}

func (c *Client) readData(id uint, reader reader, args []string) (bool, []string, error) {
	c.con.StartResponse(id)
	defer c.con.EndResponse(id)

	return reader.ReadData(c, args)
}

func (c *Client) read() ([]string, error) {
	response := []string{}

	for {
		line, err := c.readLine()
		if err != nil {
			return nil, err
		}
		if line == "OK" {
			break
		}
		response = append(response, line)
	}

	return response, nil
}

func (c *Client) readLine() (string, error) {
	line, err := c.con.ReadLine()
	if err != nil {
		return "", err
	}
	if strings.HasPrefix(line, "ACK ") {
		return "", textproto.ProtocolError(fmt.Sprintf("Unexpected line: %s", line[4:]))
	}

	return line, nil
}

func (c *Client) readBytes(bytes []byte) error {
	n, err := io.ReadFull(c.con.R, bytes)
	if err != nil {
		return err
	}
	if n != len(bytes) {
		return textproto.ProtocolError(fmt.Sprintf("Wrong byte count: %d instead of %d", n, len(bytes)))
	}

	return nil
}

func (c *Client) extractAlbum(song map[string]string) *common.Album {
	albumTitle := "Various"
	if title, titleSet := song["Album"]; titleSet {
		albumTitle = title
	}
	album := common.NewAlbum(albumTitle)

	if albumLookup, albumExists := c.albums[album.ID()]; albumExists {
		return albumLookup
	}

	return album
}

func (c *Client) extractPlaylistAlbum(song map[string]string) *common.PlaylistAlbum {
	albumTitle := "Various"
	if title, titleSet := song["Album"]; titleSet {
		albumTitle = title
	}
	album := common.NewAlbum(albumTitle)

	if albumLookup, albumExists := c.albums[album.ID()]; albumExists {
		album = albumLookup
	}

	playlistAlbum := common.NewPlaylistAlbum(album)

	return playlistAlbum
}

func (c *Client) extractSong(data map[string]string) *common.Song {
	if artist, artistSet := data["Artist"]; artistSet {
		if title, titleSet := data["Title"]; titleSet {
			if file, fileSet := data["file"]; fileSet {
				song := common.NewSong(artist, title, file)

				if track, isSet := data["Track"]; isSet {
					song.SetTrack(track)
				}
				if timeString, isSet := data["Time"]; isSet {
					time, timeErr := strconv.ParseUint(timeString, 10, 64)
					if timeErr == nil {
						song.SetLength(time)
					} else {
						log.Println("failed to parse \"Time\" field of song file", file)
					}
				}
				if date, isSet := data["Date"]; isSet {
					song.SetDate(date)
				}
				if albumartist, isSet := data["Albumartist"]; isSet {
					song.AddAlbumartist(albumartist)
				}
				if lastModifiedString, isSet := data["last-modified"]; isSet {
					lastModified, dateErr := time.Parse("2006-01-02T15:04:05-0700", lastModifiedString)
					if dateErr == nil {
						song.SetLastModified(lastModified)
					} else {
						log.Println("failed to parse \"last-modified\" field of song file", file)
					}
				}

				return song
			}
		}
	}

	return nil
}

func (c *Client) extractPlaylistSong(data map[string]string) *common.PlaylistSong {
	song := c.extractSong(data)
	if song != nil {
		if id, idSet := data["Id"]; idSet {
			if posString, posSet := data["Pos"]; posSet {
				pos, posErr := strconv.ParseUint(posString, 10, 64)
				if posErr != nil {
					log.Println("failed to parse \"Pos\" field of playlist song")
				}

				return common.NewPlaylistSong(song, id, pos)
			}
		}
	}

	return nil
}